rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-28 22:56:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: enable JSON2 columns on non-append-only tables (#9985)

Browse Source 
feat(mito2): support JSON2 in non-append-only tables

Signed-off-by: luofucong <luofc@foxmail.com>
This commit is contained in:
LFC
2026-10-28 02:43:29 +00:00
committed by GitHub
parent 3c490bd2dc
commit b369d35d8b
17 changed files with 1655 additions and 291 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mito2/src/read/compat.rs
+145 -32
View File
@@ -42,9 +42,9 @@ use crate::error::{
NewRecordBatchSnafu, Result, UnexpectedSnafu, UnsupportedOperationSnafu,
};
use crate::read::flat_projection::{FlatProjectionMapper, flat_projected_columns};
use crate::sst::internal_fields;
use crate::sst::parquet::flat_format::{FlatReadFormat, primary_key_column_index};
use crate::sst::parquet::format::{INTERNAL_COLUMN_NUM, PrimaryKeyArray};
use crate::sst::{internal_fields, tag_maybe_to_dictionary_field, with_field_id};
/// Returns true if the columns in the `projection_mapper` and `read_format` have same data types
/// and primary key encodings.
@@ -116,8 +116,12 @@ impl FlatCompatBatch {
return FlatCompatBatch::try_new_compact_sparse(mapper, actual);
}
let (index_or_defaults, fields) =
Self::compute_index_and_fields(&actual_schema, expect_schema, mapper.metadata())?;
let (index_or_defaults, fields) = Self::compute_index_and_fields(
&actual_schema,
expect_schema,
mapper.metadata(),
mapper.input_arrow_schema(compaction).fields(),
)?;
let compat_pk = FlatCompatPrimaryKey::new(mapper.metadata(), actual)?;
@@ -132,6 +136,7 @@ impl FlatCompatBatch {
actual_schema: &[(ColumnId, ConcreteDataType)],
expect_schema: &[(ColumnId, ConcreteDataType)],
expect_metadata: &RegionMetadata,
target_fields: &[FieldRef],
) -> Result<(Vec<IndexOrDefault>, Vec<FieldRef>)> {
// Maps column id to the index and data type in the actual schema.
let actual_schema_index: HashMap<_, _> = actual_schema
@@ -142,29 +147,18 @@ impl FlatCompatBatch {
let mut index_or_defaults = Vec::with_capacity(expect_schema.len());
let mut fields = Vec::with_capacity(expect_schema.len());
for (column_id, expect_data_type) in expect_schema {
for (field_index, (column_id, expect_data_type)) in expect_schema.iter().enumerate() {
// Safety: expect_schema comes from the same mapper.
let column_index = expect_metadata.column_index_by_id(*column_id).unwrap();
let expect_column = &expect_metadata.column_metadatas[column_index];
let column_field = &expect_metadata.schema.arrow_schema().fields()[column_index];
// For tag columns, we need to create a dictionary field.
if expect_column.semantic_type == SemanticType::Tag {
let field = tag_maybe_to_dictionary_field(
&expect_column.column_schema.data_type,
column_field,
);
fields.push(Arc::new(with_field_id(
(*field).clone(),
expect_column.column_id,
)));
} else {
let field = with_field_id(
Arc::unwrap_or_clone(column_field.clone()),
expect_column.column_id,
)
.with_data_type(expect_data_type.as_arrow_type());
fields.push(Arc::new(field))
};
// Query target fields follow mapper.batch_schema(); compaction reads all columns in
// PK -> fields -> timestamp order (fields -> timestamp for sparse encoding), before
// trailing internal columns. Keep this order consistent with expect_schema.
// The mapper already applies string-tag dictionary types and JSON2 compaction layouts,
// which can differ from the logical types in expect_schema.
let column_field = &target_fields[field_index];
debug_assert_eq!(column_field.name(), &expect_column.column_schema.name);
fields.push(column_field.clone());
if let Some((index, actual_data_type)) = actual_schema_index.get(column_id) {
let mut cast_type = None;
@@ -194,14 +188,13 @@ impl FlatCompatBatch {
// Create a default vector with 1 element for that column.
let default_vector = if expect_data_type.is_json2() {
// JSON2 columns do not support defaults, so missing columns use NULL.
// `expect_data_type` may be a query-concretized JSON2 type,
// while region metadata still carries its unconcretized type.
// The default vector must match the target Arrow field above.
Helper::try_into_vector(new_null_array(&expect_data_type.as_arrow_type(), 1))
.context(CreateDefaultSnafu {
// A logical JSON2 placeholder cannot describe a compaction layout.
Helper::try_into_vector(new_null_array(column_field.data_type(), 1)).context(
CreateDefaultSnafu {
region_id: expect_metadata.region_id,
column: &expect_column.column_schema.name,
})?
},
)?
} else {
expect_column
.column_schema
@@ -256,8 +249,12 @@ impl FlatCompatBatch {
.map(|col| (col.column_id, col.column_schema.data_type.clone()))
.collect();
let (index_or_defaults, fields) =
Self::compute_index_and_fields(&actual_schema, &expect_schema, mapper.metadata())?;
let (index_or_defaults, fields) = Self::compute_index_and_fields(
&actual_schema,
&expect_schema,
mapper.metadata(),
mapper.input_arrow_schema(true).fields(),
)?;
let compat_pk = FlatCompatPrimaryKey::default();
@@ -663,10 +660,15 @@ mod tests {
};
use datatypes::arrow::datatypes::UInt32Type;
use datatypes::arrow::record_batch::RecordBatch;
use datatypes::extension::json::JsonMetadata;
use datatypes::json::JsonSettings;
use datatypes::prelude::ConcreteDataType;
use datatypes::schema::ColumnSchema;
use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema};
use datatypes::types::json_type::JsonNativeType;
use datatypes::types::parse_string_to_jsonb;
use datatypes::value::ValueRef;
use datatypes::vectors::json::array::JsonArray;
use datatypes::vectors::json::json2_physical_data_type;
use mito_codec::row_converter::{
DensePrimaryKeyCodec, PrimaryKeyCodecExt, SparsePrimaryKeyCodec,
};
@@ -678,7 +680,9 @@ mod tests {
use crate::read::flat_projection::FlatProjectionMapper;
use crate::read::read_columns::ReadColumns;
use crate::sst::parquet::flat_format::FlatReadFormat;
use crate::sst::parquet::{Json2RewriteTargets, Json2TargetLayout};
use crate::sst::{FlatSchemaOptions, to_flat_sst_arrow_schema};
use crate::test_util;
/// Creates a new [RegionMetadata].
fn new_metadata(
@@ -1484,4 +1488,113 @@ mod tests {
assert_eq!(expected_batch, result);
}
#[test]
fn test_compaction_json2_uses_target_fields()
-> std::result::Result<(), Box<dyn std::error::Error>> {
let (metadata, batch) = test_util::sst_util::new_json2_record_batch()?;
let logical_settings = JsonSettings::default();
let layout = JsonSettings::try_new(vec![], Some(0))?;
let targets: Json2RewriteTargets = Arc::new(BTreeMap::from([(
0,
Json2TargetLayout {
extension_metadata: serde_json::to_string(&JsonMetadata::new(
logical_settings.clone(),
))?,
target_layout: layout.clone(),
},
)]));
// Model the existing Rewrite output; the compatibility layer only selects or fills arrays.
let json = JsonArray::from(batch.column(0)).rewrite_to_v2(
batch.schema_ref().field(0),
&logical_settings,
&layout,
)?;
let mut fields = batch.schema_ref().fields().to_vec();
fields[0] = Arc::new(
fields[0]
.as_ref()
.clone()
.with_data_type(json.data_type().clone()),
);
let mut columns = batch.columns().to_vec();
columns[0] = json;
let batch = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)?;
for encoding in [PrimaryKeyEncoding::Dense, PrimaryKeyEncoding::Sparse] {
for missing_json in [false, true] {
let source = if missing_json {
new_metadata(
&[(
1,
SemanticType::Timestamp,
ConcreteDataType::timestamp_millisecond_datatype(),
)],
&[],
)
} else {
metadata.as_ref().clone()
};
let mut builder = RegionMetadataBuilder::from_existing(source);
builder.primary_key_encoding(encoding);
let actual = Arc::new(builder.build()?);
let mut builder = RegionMetadataBuilder::from_existing(metadata.as_ref().clone());
builder.primary_key_encoding(encoding);
builder.push_column_metadata(ColumnMetadata {
column_schema: ColumnSchema::new(
"added",
ConcreteDataType::int64_datatype(),
true,
)
.with_default_constraint(Some(
ColumnDefaultConstraint::Value(Value::Int64(42)),
))?,
semantic_type: SemanticType::Field,
column_id: 2,
});
let expected = Arc::new(builder.build()?);
let read_columns = ReadColumns::new([0, 1, 2]);
let mapper = FlatProjectionMapper::new_with_json2_rewrite_targets(
&expected,
vec![0, 1, 2],
read_columns.clone(),
&targets,
)?;
let read_format = FlatReadFormat::new(actual, read_columns, None, "test", true)?;
let compat = FlatCompatBatch::try_new(&mapper, &read_format, true)?
.ok_or("missing columns require compatibility")?;
let input = if missing_json {
batch.project(&[1, 2, 3, 4])?
} else {
batch.clone()
};
let output = compat.compat(input)?;
assert_eq!(
output.column(0).data_type(),
&json2_physical_data_type(&layout)
);
assert_eq!(
output.column(0).null_count(),
if missing_json { 2 } else { 0 }
);
let defaults = output
.column(1)
.as_any()
.downcast_ref::<Int64Array>()
.ok_or("expected int64 defaults")?;
assert_eq!(defaults.values().as_ref(), &[42, 42]);
if !missing_json {
let jsonb = JsonArray::from(output.column(0)).project_to_v2(
output.schema_ref().field(0),
&datatypes::arrow::datatypes::DataType::Binary,
)?;
let jsonb = jsonb
.as_any()
.downcast_ref::<BinaryArray>()
.ok_or("expected JSONB")?;
assert_eq!(jsonb.value(0), parse_string_to_jsonb(r#"{"a":1}"#)?);
assert_eq!(jsonb.value(1), parse_string_to_jsonb(r#"{"a":2}"#)?);
}
}
}
Ok(())
}
}
src/mito2/src/read/flat_projection.rs
+57 -102
View File
@@ -17,10 +17,7 @@
use std::sync::Arc;
use api::v1::SemanticType;
use common_error::ext::BoxedError;
use common_recordbatch::error::{
ArrowComputeSnafu, DataTypesSnafu, ExternalSnafu, NewDfRecordBatchSnafu,
};
use common_recordbatch::error::{ArrowComputeSnafu, NewDfRecordBatchSnafu};
use common_recordbatch::{DfRecordBatch, RecordBatch};
use datatypes::arrow::array::{Array, AsArray};
use datatypes::arrow::datatypes::{DataType as ArrowDataType, Field};
@@ -28,15 +25,13 @@ use datatypes::extension::json::is_json2_extension_type;
use datatypes::prelude::{ConcreteDataType, DataType};
use datatypes::schema::{Schema, SchemaRef};
use datatypes::value::Value;
use datatypes::vectors::Helper;
use datatypes::vectors::json::array::JsonArray;
use datatypes::vectors::json::json2_physical_data_type;
use snafu::{OptionExt, ResultExt};
use store_api::metadata::{RegionMetadata, RegionMetadataRef};
use store_api::storage::ColumnId;
use crate::cache::CacheStrategy;
use crate::error::{InvalidRequestSnafu, RecordBatchSnafu, Result};
use crate::error::{InvalidRequestSnafu, NewRecordBatchSnafu, Result};
use crate::read::projection::{read_column_ids_from_projection, repeated_vector_with_cache};
use crate::read::read_columns::{JsonTargetTypes, ReadColumns};
use crate::sst::parquet::Json2RewriteTargets;
@@ -150,7 +145,7 @@ impl FlatProjectionMapper {
flat_projected_columns(metadata, &format_projection, read_cols.json_target_types());
let input_arrow_schema =
compute_input_arrow_schema(metadata, &batch_schema, &read_cols, json2_rewrite_targets);
compute_input_arrow_schema(metadata, &batch_schema, json2_rewrite_targets);
// If projection is empty, we don't output any column.
let output_schema = if is_empty_projection {
@@ -307,6 +302,7 @@ impl FlatProjectionMapper {
/// Converts a flat format [RecordBatch] to a normal [RecordBatch].
///
/// The batch must match the `projection` using to build the mapper.
/// JSON2 columns must already match query types before merge and deduplication.
pub(crate) fn convert(
&self,
batch: &datatypes::arrow::record_batch::RecordBatch,
@@ -365,13 +361,6 @@ impl FlatProjectionMapper {
}
}
let field = &self.output_schema.arrow_schema().fields()[output_idx];
if is_json2_extension_type(field) && array.data_type() != field.data_type() {
array = JsonArray::from(&array)
.project_to_v2(batch.schema_ref().field(*index), field.data_type())
.context(DataTypesSnafu)?;
}
arrays.push(array);
}
@@ -383,30 +372,6 @@ impl FlatProjectionMapper {
df_record_batch,
))
}
/// Projects columns from the input batch and converts them into vectors.
pub(crate) fn project_vectors(
&self,
batch: &datatypes::arrow::record_batch::RecordBatch,
) -> common_recordbatch::error::Result<Vec<datatypes::vectors::VectorRef>> {
let mut columns = Vec::with_capacity(self.output_schema.num_columns());
for index in &self.batch_indices {
let mut array = batch.column(*index).clone();
// Casts dictionary values to the target type.
if let datatypes::arrow::datatypes::DataType::Dictionary(_key_type, value_type) =
array.data_type()
{
let casted = datatypes::arrow::compute::cast(&array, value_type)
.context(ArrowComputeSnafu)?;
array = casted;
}
let vector = Helper::try_into_vector(array)
.map_err(BoxedError::new)
.context(ExternalSnafu)?;
columns.push(vector);
}
Ok(columns)
}
}
fn single_value_string_dictionary<'a>(
@@ -481,7 +446,6 @@ pub(crate) fn flat_projected_columns(
pub(crate) fn compute_input_arrow_schema(
metadata: &RegionMetadata,
batch_schema: &[(ColumnId, ConcreteDataType)],
read_cols: &ReadColumns,
json2_rewrite_targets: &Json2RewriteTargets,
) -> datatypes::arrow::datatypes::SchemaRef {
let mut new_fields = Vec::with_capacity(batch_schema.len() + 3);
@@ -489,11 +453,6 @@ pub(crate) fn compute_input_arrow_schema(
let data_type = json2_rewrite_targets
.get(column_id)
.map(|x| json2_physical_data_type(&x.target_layout))
.or_else(|| {
read_cols
.json_target_type(*column_id)
.map(|x| x.as_arrow_type())
})
.unwrap_or_else(|| data_type.as_arrow_type());
let column_metadata = metadata.column_by_id(*column_id).unwrap();
@@ -521,8 +480,7 @@ pub(crate) fn compute_input_arrow_schema(
/// Helper to project compaction batches into flat format columns
/// (fields + time index + __primary_key + __sequence + __op_type).
pub(crate) struct CompactionProjectionMapper {
mapper: FlatProjectionMapper,
assembler: DfBatchAssembler,
projection: Vec<usize>,
}
impl CompactionProjectionMapper {
@@ -544,77 +502,74 @@ impl CompactionProjectionMapper {
let read_col_ids = metadata.column_metadatas.iter().map(|col| col.column_id);
let read_cols = ReadColumns::new(read_col_ids);
let mapper = FlatProjectionMapper::new_with_read_columns(metadata, projection, read_cols)?;
let assembler = DfBatchAssembler::new(mapper.output_schema());
Ok(Self { mapper, assembler })
Ok(Self {
projection: mapper.batch_indices,
})
}
/// Projects columns and appends internal columns for compaction output.
///
/// The input batch is expected to be in flat format with internal columns appended.
pub(crate) fn project(&self, batch: DfRecordBatch) -> Result<DfRecordBatch> {
let columns = self
.mapper
.project_vectors(&batch)
.context(RecordBatchSnafu)?;
self.assembler
.build_df_record_batch_with_internal(&batch, columns)
.context(RecordBatchSnafu)
}
}
/// Builds [DfRecordBatch] with internal columns appended.
pub(crate) struct DfBatchAssembler {
output_arrow_schema_with_internal: datatypes::arrow::datatypes::SchemaRef,
}
impl DfBatchAssembler {
/// Precomputes the output schema with internal columns.
pub(crate) fn new(output_schema: SchemaRef) -> Self {
let fields = output_schema
.arrow_schema()
.fields()
.into_iter()
.chain(internal_fields().iter())
.cloned()
.collect::<Vec<_>>();
let output_arrow_schema_with_internal =
Arc::new(datatypes::arrow::datatypes::Schema::new(fields));
Self {
output_arrow_schema_with_internal,
}
}
/// Builds a [DfRecordBatch] from projected vectors plus internal columns.
///
/// Assumes the input batch already contains internal columns as the last three fields
/// ("__primary_key", "__sequence", "__op_type").
pub(crate) fn build_df_record_batch_with_internal(
&self,
batch: &datatypes::arrow::record_batch::RecordBatch,
mut columns: Vec<datatypes::vectors::VectorRef>,
) -> common_recordbatch::error::Result<DfRecordBatch> {
let num_columns = batch.columns().len();
// The last 3 columns are the internal columns.
let internal_indices = [num_columns - 3, num_columns - 2, num_columns - 1];
for index in internal_indices.iter() {
let array = batch.column(*index).clone();
let vector = Helper::try_into_vector(array)
.map_err(BoxedError::new)
.context(ExternalSnafu)?;
columns.push(vector);
}
RecordBatch::to_df_record_batch(self.output_arrow_schema_with_internal.clone(), columns)
let mut projection = self.projection.clone();
projection.extend(batch.num_columns() - 3..batch.num_columns());
// Projection must preserve the Arrow layouts already produced by compaction Rewrite.
batch.project(&projection).context(NewRecordBatchSnafu)
}
}
#[cfg(test)]
mod tests {
use datatypes::schema::ColumnSchema;
use datatypes::vectors::Helper;
use store_api::metadata::{ColumnMetadata, RegionMetadataBuilder};
use store_api::storage::RegionId;
use super::*;
use datatypes::types::json_type::JsonNativeType;
#[test]
fn test_json2_root_hint_uses_output_type_before_merge()
-> std::result::Result<(), Box<dyn std::error::Error>> {
let mut builder = RegionMetadataBuilder::new(RegionId::new(1024, 0));
builder
.push_column_metadata(ColumnMetadata {
column_schema: ColumnSchema::new(
"j",
ConcreteDataType::json2(JsonNativeType::object()),
true,
),
semantic_type: SemanticType::Field,
column_id: 0,
})
.push_column_metadata(ColumnMetadata {
column_schema: ColumnSchema::new(
"ts",
ConcreteDataType::timestamp_millisecond_datatype(),
false,
),
semantic_type: SemanticType::Timestamp,
column_id: 1,
});
let metadata = Arc::new(builder.build()?);
let read_cols = ReadColumns::new([0, 1])
.with_json_target_types([(0, JsonNativeType::String)].into_iter().collect());
let mapper = FlatProjectionMapper::new_with_read_columns(&metadata, vec![0, 1], read_cols)?;
let input = mapper.input_arrow_schema(false);
assert_eq!(
&ArrowDataType::Binary,
input.field_with_name("j")?.data_type()
);
assert_eq!(
mapper
.output_schema()
.arrow_schema()
.field_with_name("j")?
.data_type(),
input.field_with_name("j")?.data_type(),
);
Ok(())
}
fn metadata_with_legacy_json() -> RegionMetadataRef {
let mut builder = RegionMetadataBuilder::new(RegionId::new(1024, 0));
src/mito2/src/read/last_row.rs
+65 -2
View File
@@ -16,7 +16,8 @@
use std::sync::Arc;
use datatypes::arrow::array::{Array, BinaryArray, UInt32Array};
use api::v1::OpType;
use datatypes::arrow::array::{Array, BinaryArray, UInt8Array, UInt32Array};
use datatypes::arrow::compute::{concat_batches, take_record_batch};
use datatypes::arrow::record_batch::RecordBatch;
use futures::{Stream, TryStreamExt};
@@ -31,7 +32,9 @@ use crate::error::{ComputeArrowSnafu, Result};
use crate::read::read_columns::JsonTargetTypes;
use crate::read::{BoxedRecordBatchStream, timestamp_array_to_i64_slice};
use crate::sst::parquet::DEFAULT_READ_BATCH_SIZE;
use crate::sst::parquet::flat_format::{primary_key_column_index, time_index_column_index};
use crate::sst::parquet::flat_format::{
op_type_column_index, primary_key_column_index, time_index_column_index,
};
use crate::sst::parquet::format::{PrimaryKeyArray, primary_key_offsets};
use crate::sst::parquet::read_columns::ParquetReadColumns;
use crate::sst::parquet::reader::FlatRowGroupReader;
@@ -287,6 +290,31 @@ impl LastKeyState {
}
impl FlatLastTimestampSelector {
/// Keeps the latest timestamp candidates in one sorted PUT batch, including ties.
pub(crate) fn select_batch(batch: RecordBatch) -> Result<RecordBatch> {
if batch.num_rows() <= 1 {
return Ok(batch);
}
let Some(op_types) = batch
.column(op_type_column_index(batch.num_columns()))
.as_any()
.downcast_ref::<UInt8Array>()
else {
return Ok(batch);
};
if op_types.values().iter().any(|op| *op != OpType::Put as u8) {
return Ok(batch);
}
let mut selector = Self::default();
let mut pending = BatchBuffer::new();
selector.on_next(batch.clone(), &mut pending)?;
selector.finish(&mut pending)?;
if pending.num_rows == batch.num_rows() {
return Ok(batch);
}
pending.concat()
}
/// Processes the next batch and appends completed-key results into `output_buffer`.
pub(crate) fn on_next(
&mut self,
@@ -471,6 +499,41 @@ mod tests {
use super::*;
#[test]
fn test_select_batch_preserves_timestamp_ties_and_deletes()
-> std::result::Result<(), Box<dyn std::error::Error>> {
let input = new_flat_batch(
&[b"a", b"a", b"a", b"b", b"b", b"c"],
&[1, 2, 2, 1, 2, 1],
&[1, 2, 3, 4, 5, 6],
);
let expected = take_record_batch(&input, &UInt32Array::from(vec![1, 2, 4, 5]))?;
assert_eq!(
expected,
FlatLastTimestampSelector::select_batch(input.clone())?
);
let empty = input.slice(0, 0);
assert_eq!(
empty,
FlatLastTimestampSelector::select_batch(empty.clone())?
);
let mut columns = input.columns().to_vec();
columns[op_type_column_index(input.num_columns())] = Arc::new(UInt8Array::from(vec![
OpType::Delete as u8,
OpType::Put as u8,
OpType::Put as u8,
OpType::Put as u8,
OpType::Put as u8,
OpType::Put as u8,
]));
let with_delete = RecordBatch::try_new(input.schema(), columns)?;
assert_eq!(
with_delete,
FlatLastTimestampSelector::select_batch(with_delete.clone())?
);
Ok(())
}
/// Helper to build a flat format RecordBatch for testing.
fn new_flat_batch(primary_keys: &[&[u8]], timestamps: &[i64], fields: &[i64]) -> RecordBatch {
let num_rows = timestamps.len();
src/mito2/src/read/scan_util.rs
+206 -9
View File
@@ -26,7 +26,9 @@ use common_telemetry::tracing;
use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricBuilder, Time};
use datatypes::arrow::record_batch::RecordBatch;
use datatypes::timestamp::timestamp_array_to_primitive;
use futures::Stream;
#[cfg(feature = "enterprise")]
use futures::TryStreamExt;
use futures::{Stream, StreamExt};
use prometheus::IntGauge;
use smallvec::SmallVec;
use snafu::ResultExt;
@@ -40,6 +42,7 @@ use crate::metrics::{
};
use crate::read::dedup::{DedupMetrics, DedupMetricsReport};
use crate::read::flat_merge::{MergeMetrics, MergeMetricsReport};
use crate::read::last_row::FlatLastTimestampSelector;
use crate::read::pruner::PartitionPruner;
use crate::read::range::{RangeMeta, RowGroupIndex};
use crate::read::scan_region::StreamContext;
@@ -50,6 +53,7 @@ use crate::sst::index::fulltext_index::applier::FulltextIndexApplyMetrics;
use crate::sst::index::inverted_index::applier::InvertedIndexApplyMetrics;
use crate::sst::parquet::file_range::{FileRange, PreFilterMode};
use crate::sst::parquet::flat_format::{sequence_column_index, time_index_column_index};
use crate::sst::parquet::json_align;
use crate::sst::parquet::reader::{MetadataCacheMetrics, ReaderFilterMetrics, ReaderMetrics};
use crate::sst::parquet::row_group::ParquetFetchMetrics;
use crate::sst::parquet::{DEFAULT_READ_BATCH_SIZE, DEFAULT_ROW_GROUP_SIZE};
@@ -1150,6 +1154,15 @@ pub(crate) struct SeriesDistributorMetrics {
pub(crate) divider_cost: Duration,
}
/// Aligns memtable and extension batches; SST readers align before schema compatibility.
pub(crate) fn align_query_source_batch(
context: &StreamContext,
batch: RecordBatch,
) -> Result<RecordBatch> {
let query_schema = context.input.mapper.input_arrow_schema(false);
json_align::align_query_batch(batch, &query_schema)
}
/// Scans memtable ranges at `index` using flat format that returns RecordBatch.
#[tracing::instrument(
skip_all,
@@ -1187,6 +1200,39 @@ pub(crate) fn scan_flat_mem_ranges(
}
}
/// Aligns memtable JSON2 columns to query types before merge/dedup.
///
/// For append-only queries with `TimeSeriesRowSelector::LastRow { after_merge: false }`,
/// first keeps all rows tied at each series' maximum timestamp within each sorted
/// PUT-only batch, avoiding JSON2 conversion for discarded rows. Final cross-source
/// LastRow selection still runs. Batches containing deletes are not reduced.
/// This optimization uses the query's series row selector, not its merge mode;
/// non-append tables retain all input rows for deduplication and last-non-null backfill.
pub(crate) fn prepare_flat_mem_stream(
context: Arc<StreamContext>,
stream: impl Stream<Item = Result<RecordBatch>> + Send + 'static,
) -> BoxedRecordBatchStream {
if context.input.read_cols.json_target_types().is_empty() {
return Box::pin(stream);
}
let select_last = context.input.append_mode
&& matches!(
context.input.series_row_selector,
Some(TimeSeriesRowSelector::LastRow { after_merge: false })
);
Box::pin(stream.map(move |batch| {
let batch = batch?;
// Select raw candidates before decoding JSON2; final cross-source selection remains.
let batch = if select_last {
FlatLastTimestampSelector::select_batch(batch)?
} else {
batch
};
align_query_source_batch(&context, batch)
}))
}
/// Files with row count greater than this threshold can contribute to the estimation.
const SPLIT_ROW_THRESHOLD: u64 = DEFAULT_ROW_GROUP_SIZE as u64;
/// Number of series threshold for splitting batches.
@@ -1562,12 +1608,10 @@ pub fn build_flat_file_range_scan_stream(
let mapper = range.compaction_projection_mapper();
while let Some(record_batch) = reader.next_batch().await? {
let record_batch = if let Some(mapper) = mapper {
let batch = mapper.project(record_batch)?;
batch
mapper.project(record_batch)?
} else {
record_batch
};
let Some(record_batch) = filter_flat_batch_by_sequence(
record_batch,
stream_ctx.input.sequence_range,
@@ -1622,10 +1666,15 @@ pub(crate) async fn scan_flat_extension_range(
let range = context.input.extension_range(index.index);
let reader = range.flat_reader(context.as_ref(), options);
let stream = reader
.read(context, partition_metrics, index)
.read(context.clone(), partition_metrics, index)
.await
.context(crate::error::ScanExternalRangeSnafu)?;
Ok(stream)
if context.input.read_cols.json_target_types().is_empty() {
return Ok(stream);
}
Ok(Box::pin(stream.and_then(move |batch| {
std::future::ready(align_query_source_batch(&context, batch))
})))
}
pub(crate) async fn maybe_scan_flat_other_ranges(
@@ -1736,21 +1785,27 @@ pub(crate) fn split_record_batch(record_batch: RecordBatch, batches: &mut VecDeq
#[cfg(test)]
mod tests {
use arrow_select::concat::concat_batches;
use common_time::Timestamp;
use datatypes::arrow::array::{BinaryArray, TimestampMillisecondArray, UInt64Array};
use datatypes::types::json_type::JsonNativeType;
use futures::TryStreamExt;
use smallvec::{SmallVec, smallvec};
use std::sync::Arc;
use std::time::Instant;
use common_time::Timestamp;
use smallvec::{SmallVec, smallvec};
use store_api::storage::RegionId;
use super::*;
use crate::cache::CacheStrategy;
use crate::memtable::bulk::context::BulkIterContext;
use crate::memtable::bulk::part_reader::BulkPartBatchIter;
use crate::memtable::{
BoxedBatchIterator, BoxedRecordBatchIterator, IterBuilder, MemtableRange,
MemtableRangeContext, MemtableStats,
};
use crate::read::flat_projection::FlatProjectionMapper;
use crate::read::range::{MemRangeBuilder, SourceIndex};
use crate::read::read_columns::ReadColumns;
use crate::read::scan_region::ScanInput;
use crate::sst::file::{FileHandle, FileMeta};
use crate::sst::file_purger::NoopFilePurger;
@@ -1853,6 +1908,148 @@ mod tests {
}
}
#[tokio::test]
async fn test_memtable_last_row_selects_before_json2_alignment()
-> std::result::Result<(), Box<dyn std::error::Error>> {
struct BatchIter(RecordBatch, Arc<BulkIterContext>, Option<SequenceRange>);
impl IterBuilder for BatchIter {
fn build(&self, _metrics: Option<MemScanMetrics>) -> Result<BoxedBatchIterator> {
Ok(Box::new(std::iter::empty()))
}
fn is_record_batch(&self) -> bool {
true
}
fn build_record_batch(
&self,
_time_range: Option<FileTimeRange>,
metrics: Option<MemScanMetrics>,
) -> Result<BoxedRecordBatchIterator> {
Ok(Box::new(BulkPartBatchIter::from_single(
self.0.clone(),
self.1.clone(),
self.2,
1,
metrics,
)))
}
}
let (metadata, batch) = crate::test_util::sst_util::new_json2_record_batch()?;
let batch = concat_batches(&batch.schema(), &[batch.clone(), batch.slice(0, 1)])?;
let mut columns = batch.columns().to_vec();
columns[1] = Arc::new(TimestampMillisecondArray::from(vec![1, 2, 3]));
let seq_index = sequence_column_index(batch.num_columns());
columns[seq_index] = Arc::new(UInt64Array::from(vec![1, 2, 3]));
let batch = RecordBatch::try_new(batch.schema(), columns)?;
let reader_context = Arc::new(BulkIterContext::new(
metadata.clone(),
None,
None,
false,
DEFAULT_READ_BATCH_SIZE,
)?);
let time_range = (Timestamp::new_millisecond(1), Timestamp::new_millisecond(3));
let env = SchedulerEnv::new().await;
for (append, selector, sequence, count) in [
(
true,
Some(TimeSeriesRowSelector::LastRow { after_merge: false }),
None,
1,
),
(
true,
Some(TimeSeriesRowSelector::LastRow { after_merge: true }),
None,
3,
),
(
false,
Some(TimeSeriesRowSelector::LastRow { after_merge: false }),
None,
3,
),
(true, None, None, 3),
(
true,
Some(TimeSeriesRowSelector::LastRow { after_merge: false }),
Some(SequenceRange::GtLtEq { min: 0, max: 2 }),
1,
),
] {
let range_context = Arc::new(MemtableRangeContext::new(
0,
Box::new(BatchIter(batch.clone(), reader_context.clone(), sequence)),
Default::default(),
));
let stats = MemtableStats {
time_range: Some(time_range),
num_rows: 3,
num_ranges: 1,
series_count: 1,
..Default::default()
};
let range = MemtableRange::new(range_context, stats.clone());
let read_cols = ReadColumns::new([0, 1])
.with_json_target_types([(0, JsonNativeType::Variant)].into_iter().collect());
let mapper =
FlatProjectionMapper::new_with_read_columns(&metadata, vec![0, 1], read_cols)?;
let input = ScanInput::builder(env.access_layer.clone(), mapper)
.with_append_mode(append)
.with_series_row_selector(selector)
.with_sequence_range(sequence)
.with_memtables(vec![MemRangeBuilder::new(range, stats)])
.build();
let context = Arc::new(StreamContext {
input,
ranges: vec![],
query_start: Instant::now(),
});
let metrics = PartitionMetrics::new(
metadata.region_id,
0,
"test_json2",
context.query_start,
false,
&ExecutionPlanMetricsSet::new(),
);
let mut raw_stream = Box::pin(scan_flat_mem_ranges(
context.clone(),
metrics,
RowGroupIndex {
index: 0,
row_group_index: -1,
},
time_range,
));
let raw = raw_stream.try_next().await?.ok_or("expected a raw batch")?;
let expected_raw = if sequence.is_some() {
batch.slice(0, 2)
} else {
batch.clone()
};
assert_eq!(expected_raw, raw);
assert!(raw_stream.try_next().await?.is_none());
let mut stream = prepare_flat_mem_stream(context, futures::stream::iter([Ok(raw)]));
let output = stream.try_next().await?.ok_or("expected a JSON2 batch")?;
assert_eq!(count, output.num_rows());
let values = output
.column(0)
.as_any()
.downcast_ref::<BinaryArray>()
.ok_or("expected aligned JSONB")?;
assert_eq!(
datatypes::types::parse_string_to_jsonb(if sequence.is_some() {
r#"{"a":2}"#
} else {
r#"{"a":1}"#
})?,
values.value(count - 1)
);
assert!(stream.try_next().await?.is_none());
}
Ok(())
}
#[tokio::test]
async fn test_should_split_flat_batches_for_merge_uses_splittable_file_rows_per_series() {
let num_rows = SPLIT_ROW_THRESHOLD * 2;
src/mito2/src/read/seq_scan.rs
+3 -3
View File
@@ -46,8 +46,8 @@ use crate::read::range_cache::{CachedRangeReader, build_range_cache_key};
use crate::read::scan_region::{ScanInput, StreamContext};
use crate::read::scan_util::{
BATCH_SIZE_THRESHOLD, MIN_COMPACTION_SPLIT_ROWS, PartitionMetrics, PartitionMetricsList,
SplitRecordBatchStream, compute_parallel_channel_size, scan_flat_file_ranges,
scan_flat_mem_ranges, should_split_flat_batches_for_merge,
SplitRecordBatchStream, compute_parallel_channel_size, prepare_flat_mem_stream,
scan_flat_file_ranges, scan_flat_mem_ranges, should_split_flat_batches_for_merge,
};
use crate::read::stream::{ConvertBatchStream, ScanBatch, ScanBatchStream};
use crate::read::{BoxedRecordBatchStream, ScannerMetrics, scan_util};
@@ -725,7 +725,7 @@ pub(crate) async fn build_flat_sources(
*index,
range_meta.time_range,
);
ordered_sources[position] = Some(Box::pin(stream) as _);
ordered_sources[position] = Some(prepare_flat_mem_stream(stream_ctx.clone(), stream));
} else if stream_ctx.is_file_range_index(*index) {
// Common manifest-level fast-skip shared by SeqScan and UnorderedScan.
// Compaction should keep reading its selected input ranges completely.
src/mito2/src/read/series_reader.rs
+4 -3
View File
@@ -36,7 +36,7 @@ use crate::read::scan_region::StreamContext;
use crate::read::scan_util::{
PartitionMetrics, SplitRecordBatchStream, compute_average_batch_size,
compute_parallel_channel_size, filter_flat_batch_by_sequence, new_filter_metrics,
scan_flat_mem_ranges, should_split_flat_batches_for_merge,
prepare_flat_mem_stream, scan_flat_mem_ranges, should_split_flat_batches_for_merge,
};
use crate::read::seq_scan::SeqScan;
use crate::read::series_candidate::validate_metric_metadata;
@@ -411,12 +411,13 @@ async fn build_uncached_series_partition_range(
for index in range_meta.row_group_indices.iter().copied() {
if stream_ctx.is_mem_range_index(index) {
let stream = Box::pin(scan_flat_mem_ranges(
let stream = scan_flat_mem_ranges(
stream_ctx.clone(),
part_metrics.clone(),
index,
range_meta.time_range,
));
);
let stream = prepare_flat_mem_stream(stream_ctx.clone(), stream);
sources.push(filter_flat_stream_by_series(
stream,
codec.clone(),
src/mito2/src/read/unordered_scan.rs
+3 -1
View File
@@ -37,7 +37,8 @@ use crate::error::{PartitionOutOfRangeSnafu, Result};
use crate::read::pruner::{PartitionPruner, Pruner};
use crate::read::scan_region::{ScanInput, StreamContext};
use crate::read::scan_util::{
PartitionMetrics, PartitionMetricsList, scan_flat_file_ranges, scan_flat_mem_ranges,
PartitionMetrics, PartitionMetricsList, prepare_flat_mem_stream, scan_flat_file_ranges,
scan_flat_mem_ranges,
};
use crate::read::stream::{ConvertBatchStream, ScanBatch, ScanBatchStream};
use crate::read::{ScannerMetrics, scan_util};
@@ -134,6 +135,7 @@ impl UnorderedScan {
*index,
range_meta.time_range,
);
let stream = prepare_flat_mem_stream(stream_ctx.clone(), stream);
for await record_batch in stream {
yield record_batch?;
}
src/mito2/src/sst/parquet/file_range.rs
+51 -22
View File
@@ -722,10 +722,20 @@ impl RangeBase {
tag_decode_state: &mut TagDecodeState,
) -> Result<Option<BooleanBuffer>> {
let metadata = self.read_format.metadata();
// A pruned column rejects the batch without decoding any tag payloads.
if self
.filters
.iter()
// Honor skip_fields/skip_tags before inspecting MaybeFilter::Pruned: a predicate
// marked Pruned is still skipped if it must not run before merge/dedup.
// For example, after adding nullable JSON2 column j to a last_non_null table,
// an old SST may contain (key=a, ts=1, f=10) without j, while a newer row has
// (key=a, ts=1, f=NULL, j={"a":1}). WHERE j IS NOT NULL is Pruned for the old
// SST, but discarding that row here would lose f=10 needed for backfill.
// Keep it for merge/dedup and evaluate the field predicate on the merged row.
// This also applies to added non-JSON2 fields, not only JSON2 columns.
let filters = self.filters.iter().filter(|ctx| {
!(skip_fields && ctx.semantic_type() == SemanticType::Field
|| skip_tags && ctx.semantic_type() == SemanticType::Tag)
});
if filters
.clone()
.any(|ctx| matches!(ctx.filter(), MaybeFilter::Pruned))
{
return Ok(None);
@@ -757,7 +767,7 @@ impl RangeBase {
let mut mask = BooleanBuffer::new_set(input.num_rows());
// Run filter one by one and combine them result
for filter_ctx in &self.filters {
for filter_ctx in filters {
let filter = match filter_ctx.filter() {
MaybeFilter::Filter(f) => f,
// Column matches.
@@ -766,14 +776,6 @@ impl RangeBase {
MaybeFilter::Pruned => return Ok(None),
};
// Skip field filters if skip_fields is true
if skip_fields && filter_ctx.semantic_type() == SemanticType::Field {
continue;
}
if skip_tags && filter_ctx.semantic_type() == SemanticType::Tag {
continue;
}
// Get the column directly by its projected index.
// If the column is missing and it's not a tag/time column, this filter is skipped.
// Assumes the projection indices align with the input batch schema.
@@ -998,6 +1000,14 @@ impl RangeBase {
mod tests {
use std::sync::Arc;
use super::*;
use crate::read::read_columns::ReadColumns;
use crate::sst::parquet::flat_format::FlatReadFormat;
use crate::sst::{FlatSchemaOptions, to_flat_sst_arrow_schema};
use crate::test_util::sst_util::{
new_record_batch_with_custom_sequence, sst_region_metadata,
sst_region_metadata_with_encoding,
};
use datafusion_expr::{col, lit};
use datatypes::arrow::array::{
BinaryDictionaryBuilder, TimestampMillisecondArray, UInt8Array, UInt64Array,
@@ -1009,15 +1019,7 @@ mod tests {
use mito_codec::row_converter::SparsePrimaryKeyCodec;
use parquet::arrow::arrow_reader::RowSelector;
use partition::expr::col as partition_col;
use super::*;
use crate::read::read_columns::ReadColumns;
use crate::sst::parquet::flat_format::FlatReadFormat;
use crate::sst::{FlatSchemaOptions, to_flat_sst_arrow_schema};
use crate::test_util::sst_util::{
new_record_batch_with_custom_sequence, sst_region_metadata,
sst_region_metadata_with_encoding,
};
use store_api::metadata::RegionMetadataBuilder;
fn new_test_range_base(filters: Vec<SimpleFilterContext>) -> RangeBase {
let metadata: RegionMetadataRef = Arc::new(sst_region_metadata());
@@ -1333,4 +1335,31 @@ mod tests {
]);
assert_eq!(actual, expected);
}
#[test]
fn test_pruned_missing_field_respects_skip_fields()
-> std::result::Result<(), Box<dyn std::error::Error>> {
let metadata = Arc::new(sst_region_metadata());
let mut builder = RegionMetadataBuilder::from_existing(metadata.as_ref().clone());
builder.push_column_metadata(store_api::metadata::ColumnMetadata {
column_schema: ColumnSchema::new("added", ConcreteDataType::float64_datatype(), true),
semantic_type: SemanticType::Field,
column_id: 100,
});
let expected = builder.build()?;
let filter =
SimpleFilterContext::new_opt(&metadata, Some(&expected), &col("added").is_not_null())
.ok_or("expected a missing-column filter")?;
assert!(matches!(filter.filter(), MaybeFilter::Pruned));
let base = new_test_range_base(vec![filter]);
let batch = new_record_batch_with_custom_sequence(&["b", "x"], 0, 4, 1);
assert!(
base.compute_filter_mask_flat(&batch, false, false, &mut TagDecodeState::new())?
.is_none()
);
let mask = base
.compute_filter_mask_flat(&batch, true, false, &mut TagDecodeState::new())?
.ok_or("skipped field filter must preserve backfill rows")?;
assert_eq!(batch.num_rows(), mask.count_set_bits());
Ok(())
}
}
src/mito2/src/sst/parquet/json_align.rs
+73 -2
View File
@@ -12,8 +12,12 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//! Shared JSON2 conversion for Parquet streams and in-memory query batches.
//! Query types must agree before merge/dedup; compaction Rewrite retains physical layouts.
use std::collections::HashMap;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use datafusion_common::cast_column;
@@ -169,6 +173,45 @@ where
}
}
/// Aligns existing JSON2 columns to the Arrow types of matching fields in
/// `query_schema`, matching columns by name.
/// Preserves column order and skips columns absent from the batch.
pub(crate) fn align_query_batch(
batch: RecordBatch,
query_schema: &SchemaRef,
) -> Result<RecordBatch> {
let mut aligned = None;
for target in query_schema
.fields()
.iter()
.filter(|field| is_json2_extension_type(field))
{
let Ok(index) = batch.schema_ref().index_of(target.name()) else {
continue;
};
let source = batch.schema_ref().field(index);
if source.data_type() != target.data_type() {
let array = align_array(batch.column(index), source, target)?;
let (fields, columns) = aligned.get_or_insert_with(|| {
(
batch.schema_ref().fields().to_vec(),
batch.columns().to_vec(),
)
});
fields[index] = Arc::new(source.clone().with_data_type(target.data_type().clone()));
columns[index] = array;
}
}
let Some((fields, columns)) = aligned else {
return Ok(batch);
};
let schema = Arc::new(Schema::new_with_metadata(
fields,
batch.schema().metadata().clone(),
));
RecordBatch::try_new(schema, columns).context(NewRecordBatchSnafu)
}
fn align_projected_batch(
rb: RecordBatch,
output_schema: &SchemaRef,
@@ -327,6 +370,8 @@ mod tests {
use std::collections::HashMap;
use std::sync::Arc;
use super::*;
use crate::test_util;
use datatypes::arrow::array::{
Array, ArrayRef, BinaryArray, Int64Array, StringArray, StringViewArray, StructArray,
};
@@ -337,8 +382,6 @@ mod tests {
use datatypes::types::parse_string_to_jsonb;
use futures::{StreamExt, stream};
use super::*;
#[test]
fn test_aligner_resolves_json2_rewrite_settings()
-> std::result::Result<(), Box<dyn std::error::Error>> {
@@ -955,4 +998,32 @@ mod tests {
fn string_array(values: impl IntoIterator<Item = &'static str>) -> ArrayRef {
Arc::new(StringArray::from_iter_values(values))
}
#[test]
fn test_query_alignment_preserves_source_columns_until_compatibility()
-> std::result::Result<(), Box<dyn std::error::Error>> {
let (_, batch) = test_util::sst_util::new_json2_record_batch()?;
let values = int_array([10, 20]);
let mut fields = vec![Arc::new(Field::new("value", DataType::Int64, true))];
fields.extend(batch.schema_ref().fields().iter().cloned());
let mut columns = vec![values.clone()];
columns.extend_from_slice(batch.columns());
let input = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)?;
let target = schema([
Field::new("new_column", DataType::Utf8, true),
Field::new("new_json", DataType::Binary, true)
.with_extension_type(Json2ExtensionType::default()),
Field::new("j", DataType::Binary, true)
.with_extension_type(Json2ExtensionType::default()),
Field::new("value", DataType::Int64, true),
]);
let aligned = align_query_batch(input, &target)?;
assert_eq!(batch.num_columns() + 1, aligned.num_columns());
assert_eq!("value", aligned.schema_ref().field(0).name());
assert!(Arc::ptr_eq(&values, aligned.column(0)));
assert_eq!("j", aligned.schema_ref().field(1).name());
assert_eq!(&DataType::Binary, aligned.column(1).data_type());
let repeated = align_query_batch(aligned.clone(), &target)?;
assert!(Arc::ptr_eq(aligned.column(1), repeated.column(1)));
Ok(())
}
}
src/mito2/src/sst/parquet/reader.rs
+8
View File
@@ -655,6 +655,14 @@ impl ParquetReaderBuilder {
let json2_rewrite_targets = self
.json2_rewrite_targets
.iter()
// Global compaction targets may include JSON2 columns added after this SST was written.
// Rewrite only projected columns present in this SST; schema compatibility fills
// missing columns with NULL later, without looking them up in this SST's metadata.
.filter(|(column_id, _)| {
read_format
.parquet_projected_index_by_id(**column_id)
.is_some()
})
.map(|(column_id, layout)| {
let column = region_meta
.column_by_id(*column_id)
src/mito2/src/test_util/sst_util.rs
+57
View File
@@ -24,10 +24,16 @@ use datatypes::arrow::array::{
ArrayRef, BinaryDictionaryBuilder, RecordBatch, StringDictionaryBuilder,
TimestampMillisecondArray, UInt8Array, UInt64Array,
};
use datatypes::arrow::array::{Int64Array, StructArray};
use datatypes::arrow::datatypes::UInt32Type;
use datatypes::arrow::datatypes::{DataType, Field};
use datatypes::extension::json::Json2ExtensionType;
use datatypes::json::JsonSettings;
use datatypes::prelude::ConcreteDataType;
use datatypes::schema::{ColumnSchema, SkippingIndexOptions};
use datatypes::types::json_type::JsonNativeType;
use datatypes::value::ValueRef;
use datatypes::vectors::json::array::JsonArray;
use mito_codec::row_converter::{DensePrimaryKeyCodec, PrimaryKeyCodecExt, SortField};
use object_store::layers::mock::{self, MockLayer, MockLayerBuilder, oio};
use store_api::metadata::{
@@ -44,6 +50,57 @@ use crate::sst::file::{FileHandle, FileMeta};
use crate::sst::{FlatSchemaOptions, to_flat_sst_arrow_schema};
use crate::test_util::{new_batch_builder, new_noop_file_purger};
/// Builds a current-layout JSON2 flat batch with two timestamps in one series.
pub(crate) fn new_json2_record_batch()
-> Result<(RegionMetadataRef, RecordBatch), Box<dyn std::error::Error>> {
let legacy: ArrayRef = Arc::new(StructArray::from(vec![(
Arc::new(Field::new("a", DataType::Int64, true)),
Arc::new(Int64Array::from_iter_values([1, 2])) as ArrayRef,
)]));
let field = Field::new("j", legacy.data_type().clone(), true)
.with_extension_type(Json2ExtensionType::default());
let settings = JsonSettings::default();
let physical = JsonArray::from(&legacy).rewrite_to_v2(&field, &settings, &settings)?;
let mut column_schema =
ColumnSchema::new("j", ConcreteDataType::json2(JsonNativeType::object()), true);
column_schema.with_extension_type(&Json2ExtensionType::default());
let mut metadata = RegionMetadataBuilder::new(RegionId::new(1024, 0));
metadata
.push_column_metadata(ColumnMetadata {
column_schema,
semantic_type: SemanticType::Field,
column_id: 0,
})
.push_column_metadata(ColumnMetadata {
column_schema: ColumnSchema::new(
"ts",
ConcreteDataType::timestamp_millisecond_datatype(),
false,
),
semantic_type: SemanticType::Timestamp,
column_id: 1,
});
let metadata = Arc::new(metadata.build()?);
let mut options = FlatSchemaOptions::default();
options
.concretized_json_types
.insert("j".to_string(), physical.data_type().clone());
let mut keys = BinaryDictionaryBuilder::<UInt32Type>::new();
keys.append(b"series")?;
keys.append(b"series")?;
let input = RecordBatch::try_new(
to_flat_sst_arrow_schema(&metadata, &options),
vec![
physical,
Arc::new(TimestampMillisecondArray::from(vec![1, 2])),
Arc::new(keys.finish()),
Arc::new(UInt64Array::from_value(1, 2)),
Arc::new(UInt8Array::from_value(OpType::Put as u8, 2)),
],
)?;
Ok((metadata, input))
}
/// Test region id.
const REGION_ID: RegionId = RegionId::new(0, 0);
src/operator/src/statement/ddl.rs
+37 -38
View File
@@ -100,7 +100,6 @@ use sqlparser::ast::{
use store_api::metric_engine_consts::{
LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY,
};
use store_api::mito_engine_options::APPEND_MODE_KEY;
use substrait::{DFLogicalSubstraitConvertor, SubstraitPlan};
use table::TableRef;
use table::dist_table::DistTable;
@@ -2632,7 +2631,7 @@ pub fn verify_alter(
.build()
.context(error::BuildTableMetaSnafu { table_name })?;
validate_json2_columns_append_mode(&new_meta.schema, &new_meta.options)?;
validate_json2_primary_keys(&new_meta.schema, &new_meta.primary_key_indices)?;
Ok(true)
}
@@ -2682,7 +2681,7 @@ pub fn create_table_info(
let mut table_options = TableOptions::try_from_iter(&create_table.table_options)
.context(UnrecognizedTableOptionSnafu)?;
validate_json2_columns_append_mode(&schema, &table_options)?;
validate_json2_primary_keys(&schema, &primary_key_indices)?;
validate_and_normalize_annotations(&mut table_options, &schema, &partition_key_indices)?;
@@ -2721,26 +2720,16 @@ pub fn create_table_info(
Ok(table_info)
}
fn validate_json2_columns_append_mode(schema: &Schema, table_options: &TableOptions) -> Result<()> {
let append_mode = table_options
.extra_options
.get(APPEND_MODE_KEY)
.is_some_and(|value| value == "true");
for column in schema.column_schemas() {
if column.data_type.is_json2() {
ensure!(
append_mode,
InvalidSqlSnafu {
err_msg: format!(
"JSON2 column `{}` requires {}='true'",
column.name, APPEND_MODE_KEY
),
}
);
}
fn validate_json2_primary_keys(schema: &Schema, primary_key_indices: &[usize]) -> Result<()> {
for &index in primary_key_indices {
let column = &schema.column_schemas()[index];
ensure!(
!column.data_type.is_json2(),
InvalidSqlSnafu {
err_msg: format!("JSON2 column `{}` cannot be a primary key", column.name),
}
);
}
Ok(())
}
@@ -4147,26 +4136,36 @@ WITH ('repartition.column.hint' = ' host ')",
}
#[test]
fn test_json2_requires_append_mode() {
let cases = [
"CREATE TABLE monitor (payload JSON2, ts TIMESTAMP TIME INDEX);",
"CREATE TABLE monitor (payload JSON2, ts TIMESTAMP TIME INDEX) WITH (append_mode='false');",
];
for sql in cases {
let expr = create_expr_from_sql(sql);
let err = create_table_info(&expr, vec![]).unwrap_err();
assert!(
err.to_string()
.contains("JSON2 column `payload` requires append_mode='true'"),
"{err}"
);
fn test_json2_supports_non_append_mode() -> std::result::Result<(), Box<dyn std::error::Error>>
{
for options in [
"",
"WITH (append_mode='false')",
"WITH (append_mode='true')",
"WITH (merge_mode='last_non_null')",
] {
let expr = create_expr_from_sql(&format!(
"CREATE TABLE monitor (payload JSON2, ts TIMESTAMP TIME INDEX) {options};"
));
create_table_info(&expr, vec![])?;
}
Ok(())
}
#[test]
fn test_json2_cannot_be_primary_key() -> std::result::Result<(), Box<dyn std::error::Error>> {
let expr = create_expr_from_sql(
"CREATE TABLE monitor (payload JSON2, ts TIMESTAMP TIME INDEX) WITH (append_mode='true');",
"CREATE TABLE monitor (payload JSON2, ts TIMESTAMP TIME INDEX, PRIMARY KEY(payload));",
);
create_table_info(&expr, vec![]).unwrap();
let err = create_table_info(&expr, vec![])
.err()
.ok_or("expected JSON2 primary key rejection")?;
assert!(
err.to_string()
.contains("JSON2 column `payload` cannot be a primary key"),
"{err}"
);
Ok(())
}
#[test]
src/query/src/datafusion.rs
+35 -26
View File
@@ -233,14 +233,47 @@ impl DatafusionQueryEngine {
}
}
WriteOp::Delete => {
// Read-only system tables may have no time index; reject DELETE before key lookup.
ensure!(
!is_readonly_table(&table_name.schema, &table_name.table),
TableReadOnlySnafu {
table: table_name.table.to_string(),
}
);
let table_info = table.table_info();
let table_schema = table.schema();
let ts_column = table_schema.timestamp_column().with_context(|| {
MissingTimestampColumnSnafu {
table_name: table_info.name.clone(),
}
})?;
let key_indices = stream
.schema()
.column_schemas()
.iter()
.enumerate()
.filter(|(_, column)| {
ts_column.name == column.name
|| table_info
.meta
.row_key_column_names()
.any(|name| name == &column.name)
})
.map(|(index, _)| index)
.collect::<Vec<_>>();
while let Some(batch) = stream.next().await {
let batch = batch.context(CreateRecordBatchSnafu)?;
// The input plan already evaluated WHERE; storage deletes need only keys/time.
// Project before vector conversion because JSON2's logical Struct is a placeholder.
let batch = batch
.context(CreateRecordBatchSnafu)?
.try_project(&key_indices)
.context(CreateRecordBatchSnafu)?;
let column_vectors = batch
.column_vectors(&table_name.to_string(), table.schema())
.map_err(BoxedError::new)
.context(QueryExecutionSnafu)?;
affected_rows += self
.delete(&table_name, &table, column_vectors, query_ctx.clone())
.delete(&table_name, column_vectors, query_ctx.clone())
.await?;
}
}
@@ -256,36 +289,12 @@ impl DatafusionQueryEngine {
async fn delete(
&self,
table_name: &ResolvedTableReference,
table: &TableRef,
column_vectors: HashMap<String, VectorRef>,
query_ctx: QueryContextRef,
) -> Result<usize> {
let catalog_name = table_name.catalog.to_string();
let schema_name = table_name.schema.to_string();
let table_name = table_name.table.to_string();
let table_schema = table.schema();
ensure!(
!is_readonly_table(&schema_name, &table_name),
TableReadOnlySnafu { table: table_name }
);
let ts_column = table_schema
.timestamp_column()
.map(|x| &x.name)
.with_context(|| MissingTimestampColumnSnafu {
table_name: table_name.clone(),
})?;
let table_info = table.table_info();
let rowkey_columns = table_info
.meta
.row_key_column_names()
.collect::<Vec<&String>>();
let column_vectors = column_vectors
.into_iter()
.filter(|x| &x.0 == ts_column || rowkey_columns.contains(&&x.0))
.collect::<HashMap<_, _>>();
let request = DeleteRequest {
catalog_name,
tests/cases/standalone/common/types/json/json2_limit.result
+22 -31
View File
@@ -65,36 +65,6 @@ drop table json2_whole_and_path_read;
Affected Rows: 0
create table json2_without_append_mode (
ts timestamp time index,
j json2
);
Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` requires append_mode='true'
create table json2_append_mode_false (
ts timestamp time index,
j json2
) with (
'append_mode' = 'false'
);
Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` requires append_mode='true'
create table json2_alter_non_append (
ts timestamp time index
);
Affected Rows: 0
alter table json2_alter_non_append add column j json2;
Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` requires append_mode='true'
drop table json2_alter_non_append;
Affected Rows: 0
create table json2_set_append_mode_false (
ts timestamp time index,
j json2
@@ -104,14 +74,35 @@ create table json2_set_append_mode_false (
Affected Rows: 0
-- SQLNESS REPLACE \d+\(\d+,\s+\d+\) REDACTED
alter table json2_set_append_mode_false set 'append_mode' = 'false';
Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` requires append_mode='true'
Error: 1004(InvalidArguments), Invalid region request, region_id: REDACTED, err: Only allow changing append_mode from false to true
drop table json2_set_append_mode_false;
Affected Rows: 0
create table json2_primary_key (
ts timestamp time index,
j json2,
primary key (j)
);
Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` cannot be a primary key
create table json2_add_primary_key (ts timestamp time index);
Affected Rows: 0
alter table json2_add_primary_key add column j json2 primary key;
Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` cannot be a primary key
drop table json2_add_primary_key;
Affected Rows: 0
-- JSON2 requires flat SST output even when append mode is enabled.
create table json2_primary_key_format (
ts timestamp time index,
tests/cases/standalone/common/types/json/json2_limit.sql
+13 -20
View File
@@ -38,26 +38,6 @@ order by json_get(j, 'a.b');
drop table json2_whole_and_path_read;
create table json2_without_append_mode (
ts timestamp time index,
j json2
);
create table json2_append_mode_false (
ts timestamp time index,
j json2
) with (
'append_mode' = 'false'
);
create table json2_alter_non_append (
ts timestamp time index
);
alter table json2_alter_non_append add column j json2;
drop table json2_alter_non_append;
create table json2_set_append_mode_false (
ts timestamp time index,
j json2
@@ -65,10 +45,23 @@ create table json2_set_append_mode_false (
'append_mode' = 'true'
);
-- SQLNESS REPLACE \d+\(\d+,\s+\d+\) REDACTED
alter table json2_set_append_mode_false set 'append_mode' = 'false';
drop table json2_set_append_mode_false;
create table json2_primary_key (
ts timestamp time index,
j json2,
primary key (j)
);
create table json2_add_primary_key (ts timestamp time index);
alter table json2_add_primary_key add column j json2 primary key;
drop table json2_add_primary_key;
-- JSON2 requires flat SST output even when append mode is enabled.
create table json2_primary_key_format (
ts timestamp time index,
tests/cases/standalone/common/types/json/json2_non_append.result
+675
View File
@@ -0,0 +1,675 @@
create table json2_without_append_mode (
ts timestamp time index,
j json2
);
Affected Rows: 0
drop table json2_without_append_mode;
Affected Rows: 0
create table json2_append_mode_false (
ts timestamp time index,
j json2
) with (
'append_mode' = 'false'
);
Affected Rows: 0
drop table json2_append_mode_false;
Affected Rows: 0
create table json2_alter_non_append (
ts timestamp time index
);
Affected Rows: 0
insert into json2_alter_non_append values (1);
Affected Rows: 1
alter table json2_alter_non_append add column j json2;
Affected Rows: 0
insert into json2_alter_non_append values (1, '{"a":1}');
Affected Rows: 1
select ts, j.a from json2_alter_non_append;
+-------------------------+------------------------------------------------+
| ts | json_get(json2_alter_non_append.j,Utf8("$.a")) |
+-------------------------+------------------------------------------------+
| 1970-01-01T00:00:00.001 | 1 |
+-------------------------+------------------------------------------------+
drop table json2_alter_non_append;
Affected Rows: 0
-- JSON2 is merged as a whole field, including empty objects and explicit JSON nulls.
CREATE TABLE json2_last_row (
host STRING,
ts TIMESTAMP TIME INDEX,
j JSON2,
k JSON2(max_auto_expanded_paths = 0),
f DOUBLE,
PRIMARY KEY (host)
) WITH (
'append_mode' = 'false',
'merge_mode' = 'last_row'
);
Affected Rows: 0
INSERT INTO json2_last_row VALUES
('one', 1, '{"a":1}', '{"a":1}', 10),
('one', 2, '{"a":2}', '{"a":2}', 10),
('one', 3, '{"a":3}', '{"a":3}', 10),
('one', 4, '{"a":4}', '{"a":4}', 10),
('one', 5, '{"a":{"x":5}}', '{"a":{"x":5}}', 10),
('one', 6, '{"a":{"x":6}}', '{"a":{"x":6}}', 10),
('one', 7, '{"a":{"x":8}}', '{"a":{"x":8}}', 10),
('two', 1, '{"a":99}', '{"a":99}', 99);
Affected Rows: 8
ADMIN FLUSH_TABLE('json2_last_row');
+-------------------------------------+
| ADMIN FLUSH_TABLE('json2_last_row') |
+-------------------------------------+
| 0 |
+-------------------------------------+
-- New objects replace old objects; only SQL NULL can fall back in last_non_null.
INSERT INTO json2_last_row VALUES
('one', 1, '{"b":2}', '{"b":2}', NULL),
('one', 2, '{}', '{}', NULL),
('one', 3, '{"a":null}', '{"a":null}', NULL),
('one', 4, NULL, NULL, NULL),
('one', 5, '{"a":42}', '{"a":42}', NULL),
('one', 6, '{"a":{"x":7}}', '{"a":{"x":7}}', NULL),
('one', 7, NULL, NULL, NULL);
Affected Rows: 7
-- Path-only reads must keep the new nested value and distinguish SQL NULL fallback.
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_row ORDER BY host, ts;
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| host | ts | j | a | ax | k | ka | f |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | |
| one | 1970-01-01T00:00:00.002 | {} | | | {} | | |
| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | |
| one | 1970-01-01T00:00:00.004 | | | | | | |
| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | |
| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | |
| one | 1970-01-01T00:00:00.007 | | | | | | |
| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts;
+------+-------------------------+----+-----+
| host | ts | ax | kax |
+------+-------------------------+----+-----+
| one | 1970-01-01T00:00:00.001 | | |
| one | 1970-01-01T00:00:00.002 | | |
| one | 1970-01-01T00:00:00.003 | | |
| one | 1970-01-01T00:00:00.004 | | |
| one | 1970-01-01T00:00:00.005 | | |
| one | 1970-01-01T00:00:00.006 | 7 | 7 |
| one | 1970-01-01T00:00:00.007 | | |
| two | 1970-01-01T00:00:00.001 | | |
+------+-------------------------+----+-----+
SELECT host, ts FROM json2_last_row WHERE j.a = '1' ORDER BY host, ts;
+------+----+
| host | ts |
+------+----+
+------+----+
ADMIN FLUSH_TABLE('json2_last_row');
+-------------------------------------+
| ADMIN FLUSH_TABLE('json2_last_row') |
+-------------------------------------+
| 0 |
+-------------------------------------+
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_row ORDER BY host, ts;
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| host | ts | j | a | ax | k | ka | f |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | |
| one | 1970-01-01T00:00:00.002 | {} | | | {} | | |
| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | |
| one | 1970-01-01T00:00:00.004 | | | | | | |
| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | |
| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | |
| one | 1970-01-01T00:00:00.007 | | | | | | |
| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts;
+------+-------------------------+----+-----+
| host | ts | ax | kax |
+------+-------------------------+----+-----+
| one | 1970-01-01T00:00:00.001 | | |
| one | 1970-01-01T00:00:00.002 | | |
| one | 1970-01-01T00:00:00.003 | | |
| one | 1970-01-01T00:00:00.004 | | |
| one | 1970-01-01T00:00:00.005 | | |
| one | 1970-01-01T00:00:00.006 | 7 | 7 |
| one | 1970-01-01T00:00:00.007 | | |
| two | 1970-01-01T00:00:00.001 | | |
+------+-------------------------+----+-----+
ADMIN COMPACT_TABLE('json2_last_row');
+---------------------------------------+
| ADMIN COMPACT_TABLE('json2_last_row') |
+---------------------------------------+
| 0 |
+---------------------------------------+
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_row ORDER BY host, ts;
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| host | ts | j | a | ax | k | ka | f |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | |
| one | 1970-01-01T00:00:00.002 | {} | | | {} | | |
| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | |
| one | 1970-01-01T00:00:00.004 | | | | | | |
| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | |
| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | |
| one | 1970-01-01T00:00:00.007 | | | | | | |
| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts;
+------+-------------------------+----+-----+
| host | ts | ax | kax |
+------+-------------------------+----+-----+
| one | 1970-01-01T00:00:00.001 | | |
| one | 1970-01-01T00:00:00.002 | | |
| one | 1970-01-01T00:00:00.003 | | |
| one | 1970-01-01T00:00:00.004 | | |
| one | 1970-01-01T00:00:00.005 | | |
| one | 1970-01-01T00:00:00.006 | 7 | 7 |
| one | 1970-01-01T00:00:00.007 | | |
| two | 1970-01-01T00:00:00.001 | | |
+------+-------------------------+----+-----+
-- SQLNESS ARG restart=true
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_row ORDER BY host, ts;
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| host | ts | j | a | ax | k | ka | f |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | |
| one | 1970-01-01T00:00:00.002 | {} | | | {} | | |
| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | |
| one | 1970-01-01T00:00:00.004 | | | | | | |
| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | |
| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | |
| one | 1970-01-01T00:00:00.007 | | | | | | |
| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts;
+------+-------------------------+----+-----+
| host | ts | ax | kax |
+------+-------------------------+----+-----+
| one | 1970-01-01T00:00:00.001 | | |
| one | 1970-01-01T00:00:00.002 | | |
| one | 1970-01-01T00:00:00.003 | | |
| one | 1970-01-01T00:00:00.004 | | |
| one | 1970-01-01T00:00:00.005 | | |
| one | 1970-01-01T00:00:00.006 | 7 | 7 |
| one | 1970-01-01T00:00:00.007 | | |
| two | 1970-01-01T00:00:00.001 | | |
+------+-------------------------+----+-----+
-- A delete must prevent last_non_null from filling fields from older versions.
DELETE FROM json2_last_row WHERE host = 'one' AND ts = 1;
Affected Rows: 1
INSERT INTO json2_last_row VALUES ('one', 1, '{"c":3}', '{"c":3}', NULL);
Affected Rows: 1
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_row ORDER BY host, ts;
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| host | ts | j | a | ax | k | ka | f |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"c":3} | | | {"c":3} | | |
| one | 1970-01-01T00:00:00.002 | {} | | | {} | | |
| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | |
| one | 1970-01-01T00:00:00.004 | | | | | | |
| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | |
| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | |
| one | 1970-01-01T00:00:00.007 | | | | | | |
| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
ADMIN FLUSH_TABLE('json2_last_row');
+-------------------------------------+
| ADMIN FLUSH_TABLE('json2_last_row') |
+-------------------------------------+
| 0 |
+-------------------------------------+
ADMIN COMPACT_TABLE('json2_last_row');
+---------------------------------------+
| ADMIN COMPACT_TABLE('json2_last_row') |
+---------------------------------------+
| 0 |
+---------------------------------------+
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_row ORDER BY host, ts;
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| host | ts | j | a | ax | k | ka | f |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"c":3} | | | {"c":3} | | |
| one | 1970-01-01T00:00:00.002 | {} | | | {} | | |
| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | |
| one | 1970-01-01T00:00:00.004 | | | | | | |
| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | |
| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | |
| one | 1970-01-01T00:00:00.007 | | | | | | |
| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
DELETE FROM json2_last_row WHERE j.a = '99';
Affected Rows: 1
SELECT COUNT(*) FROM json2_last_row WHERE host = 'two';
+----------+
| count(*) |
+----------+
| 0 |
+----------+
DROP TABLE json2_last_row;
Affected Rows: 0
-- JSON2 is merged as a whole field, including empty objects and explicit JSON nulls.
CREATE TABLE json2_last_non_null (
host STRING,
ts TIMESTAMP TIME INDEX,
j JSON2,
k JSON2(max_auto_expanded_paths = 0),
f DOUBLE,
PRIMARY KEY (host)
) WITH (
'append_mode' = 'false',
'merge_mode' = 'last_non_null'
);
Affected Rows: 0
INSERT INTO json2_last_non_null VALUES
('one', 1, '{"a":1}', '{"a":1}', 10),
('one', 2, '{"a":2}', '{"a":2}', 10),
('one', 3, '{"a":3}', '{"a":3}', 10),
('one', 4, '{"a":4}', '{"a":4}', 10),
('one', 5, '{"a":{"x":5}}', '{"a":{"x":5}}', 10),
('one', 6, '{"a":{"x":6}}', '{"a":{"x":6}}', 10),
('one', 7, '{"a":{"x":8}}', '{"a":{"x":8}}', 10),
('two', 1, '{"a":99}', '{"a":99}', 99);
Affected Rows: 8
ADMIN FLUSH_TABLE('json2_last_non_null');
+------------------------------------------+
| ADMIN FLUSH_TABLE('json2_last_non_null') |
+------------------------------------------+
| 0 |
+------------------------------------------+
-- New objects replace old objects; only SQL NULL can fall back in last_non_null.
INSERT INTO json2_last_non_null VALUES
('one', 1, '{"b":2}', '{"b":2}', NULL),
('one', 2, '{}', '{}', NULL),
('one', 3, '{"a":null}', '{"a":null}', NULL),
('one', 4, NULL, NULL, NULL),
('one', 5, '{"a":42}', '{"a":42}', NULL),
('one', 6, '{"a":{"x":7}}', '{"a":{"x":7}}', NULL),
('one', 7, NULL, NULL, NULL);
Affected Rows: 7
-- Path-only reads must keep the new nested value and distinguish SQL NULL fallback.
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_non_null ORDER BY host, ts;
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| host | ts | j | a | ax | k | ka | f |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | 10.0 |
| one | 1970-01-01T00:00:00.002 | {} | | | {} | | 10.0 |
| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | 10.0 |
| one | 1970-01-01T00:00:00.004 | {"a":4} | 4 | | {"a":4} | 4 | 10.0 |
| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | 10.0 |
| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | 10.0 |
| one | 1970-01-01T00:00:00.007 | {"a":{"x":8}} | {"x":8} | 8 | {"a":{"x":8}} | {"x":8} | 10.0 |
| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts;
+------+-------------------------+----+-----+
| host | ts | ax | kax |
+------+-------------------------+----+-----+
| one | 1970-01-01T00:00:00.001 | | |
| one | 1970-01-01T00:00:00.002 | | |
| one | 1970-01-01T00:00:00.003 | | |
| one | 1970-01-01T00:00:00.004 | | |
| one | 1970-01-01T00:00:00.005 | | |
| one | 1970-01-01T00:00:00.006 | 7 | 7 |
| one | 1970-01-01T00:00:00.007 | 8 | 8 |
| two | 1970-01-01T00:00:00.001 | | |
+------+-------------------------+----+-----+
SELECT host, ts FROM json2_last_non_null WHERE j.a = '1' ORDER BY host, ts;
+------+----+
| host | ts |
+------+----+
+------+----+
ADMIN FLUSH_TABLE('json2_last_non_null');
+------------------------------------------+
| ADMIN FLUSH_TABLE('json2_last_non_null') |
+------------------------------------------+
| 0 |
+------------------------------------------+
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_non_null ORDER BY host, ts;
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| host | ts | j | a | ax | k | ka | f |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | 10.0 |
| one | 1970-01-01T00:00:00.002 | {} | | | {} | | 10.0 |
| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | 10.0 |
| one | 1970-01-01T00:00:00.004 | {"a":4} | 4 | | {"a":4} | 4 | 10.0 |
| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | 10.0 |
| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | 10.0 |
| one | 1970-01-01T00:00:00.007 | {"a":{"x":8}} | {"x":8} | 8 | {"a":{"x":8}} | {"x":8} | 10.0 |
| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts;
+------+-------------------------+----+-----+
| host | ts | ax | kax |
+------+-------------------------+----+-----+
| one | 1970-01-01T00:00:00.001 | | |
| one | 1970-01-01T00:00:00.002 | | |
| one | 1970-01-01T00:00:00.003 | | |
| one | 1970-01-01T00:00:00.004 | | |
| one | 1970-01-01T00:00:00.005 | | |
| one | 1970-01-01T00:00:00.006 | 7 | 7 |
| one | 1970-01-01T00:00:00.007 | 8 | 8 |
| two | 1970-01-01T00:00:00.001 | | |
+------+-------------------------+----+-----+
ADMIN COMPACT_TABLE('json2_last_non_null');
+--------------------------------------------+
| ADMIN COMPACT_TABLE('json2_last_non_null') |
+--------------------------------------------+
| 0 |
+--------------------------------------------+
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_non_null ORDER BY host, ts;
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| host | ts | j | a | ax | k | ka | f |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | 10.0 |
| one | 1970-01-01T00:00:00.002 | {} | | | {} | | 10.0 |
| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | 10.0 |
| one | 1970-01-01T00:00:00.004 | {"a":4} | 4 | | {"a":4} | 4 | 10.0 |
| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | 10.0 |
| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | 10.0 |
| one | 1970-01-01T00:00:00.007 | {"a":{"x":8}} | {"x":8} | 8 | {"a":{"x":8}} | {"x":8} | 10.0 |
| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts;
+------+-------------------------+----+-----+
| host | ts | ax | kax |
+------+-------------------------+----+-----+
| one | 1970-01-01T00:00:00.001 | | |
| one | 1970-01-01T00:00:00.002 | | |
| one | 1970-01-01T00:00:00.003 | | |
| one | 1970-01-01T00:00:00.004 | | |
| one | 1970-01-01T00:00:00.005 | | |
| one | 1970-01-01T00:00:00.006 | 7 | 7 |
| one | 1970-01-01T00:00:00.007 | 8 | 8 |
| two | 1970-01-01T00:00:00.001 | | |
+------+-------------------------+----+-----+
-- SQLNESS ARG restart=true
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_non_null ORDER BY host, ts;
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| host | ts | j | a | ax | k | ka | f |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | 10.0 |
| one | 1970-01-01T00:00:00.002 | {} | | | {} | | 10.0 |
| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | 10.0 |
| one | 1970-01-01T00:00:00.004 | {"a":4} | 4 | | {"a":4} | 4 | 10.0 |
| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | 10.0 |
| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | 10.0 |
| one | 1970-01-01T00:00:00.007 | {"a":{"x":8}} | {"x":8} | 8 | {"a":{"x":8}} | {"x":8} | 10.0 |
| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts;
+------+-------------------------+----+-----+
| host | ts | ax | kax |
+------+-------------------------+----+-----+
| one | 1970-01-01T00:00:00.001 | | |
| one | 1970-01-01T00:00:00.002 | | |
| one | 1970-01-01T00:00:00.003 | | |
| one | 1970-01-01T00:00:00.004 | | |
| one | 1970-01-01T00:00:00.005 | | |
| one | 1970-01-01T00:00:00.006 | 7 | 7 |
| one | 1970-01-01T00:00:00.007 | 8 | 8 |
| two | 1970-01-01T00:00:00.001 | | |
+------+-------------------------+----+-----+
-- A delete must prevent last_non_null from filling fields from older versions.
DELETE FROM json2_last_non_null WHERE host = 'one' AND ts = 1;
Affected Rows: 1
INSERT INTO json2_last_non_null VALUES ('one', 1, '{"c":3}', '{"c":3}', NULL);
Affected Rows: 1
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_non_null ORDER BY host, ts;
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| host | ts | j | a | ax | k | ka | f |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"c":3} | | | {"c":3} | | |
| one | 1970-01-01T00:00:00.002 | {} | | | {} | | 10.0 |
| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | 10.0 |
| one | 1970-01-01T00:00:00.004 | {"a":4} | 4 | | {"a":4} | 4 | 10.0 |
| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | 10.0 |
| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | 10.0 |
| one | 1970-01-01T00:00:00.007 | {"a":{"x":8}} | {"x":8} | 8 | {"a":{"x":8}} | {"x":8} | 10.0 |
| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
ADMIN FLUSH_TABLE('json2_last_non_null');
+------------------------------------------+
| ADMIN FLUSH_TABLE('json2_last_non_null') |
+------------------------------------------+
| 0 |
+------------------------------------------+
ADMIN COMPACT_TABLE('json2_last_non_null');
+--------------------------------------------+
| ADMIN COMPACT_TABLE('json2_last_non_null') |
+--------------------------------------------+
| 0 |
+--------------------------------------------+
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_non_null ORDER BY host, ts;
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| host | ts | j | a | ax | k | ka | f |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"c":3} | | | {"c":3} | | |
| one | 1970-01-01T00:00:00.002 | {} | | | {} | | 10.0 |
| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | 10.0 |
| one | 1970-01-01T00:00:00.004 | {"a":4} | 4 | | {"a":4} | 4 | 10.0 |
| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | 10.0 |
| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | 10.0 |
| one | 1970-01-01T00:00:00.007 | {"a":{"x":8}} | {"x":8} | 8 | {"a":{"x":8}} | {"x":8} | 10.0 |
| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 |
+------+-------------------------+---------------+---------+----+---------------+---------+------+
DELETE FROM json2_last_non_null WHERE j.a = '99';
Affected Rows: 1
SELECT COUNT(*) FROM json2_last_non_null WHERE host = 'two';
+----------+
| count(*) |
+----------+
| 0 |
+----------+
DROP TABLE json2_last_non_null;
Affected Rows: 0
-- A predicate on an added JSON2 field must retain older values of other fields.
CREATE TABLE json2_alter_backfill (
host STRING,
ts TIMESTAMP TIME INDEX,
f DOUBLE,
PRIMARY KEY (host)
) WITH ('merge_mode' = 'last_non_null');
Affected Rows: 0
INSERT INTO json2_alter_backfill VALUES ('one', 1, 10);
Affected Rows: 1
ADMIN FLUSH_TABLE('json2_alter_backfill');
+-------------------------------------------+
| ADMIN FLUSH_TABLE('json2_alter_backfill') |
+-------------------------------------------+
| 0 |
+-------------------------------------------+
ALTER TABLE json2_alter_backfill ADD COLUMN j JSON2;
Affected Rows: 0
INSERT INTO json2_alter_backfill VALUES ('one', 1, NULL, '{"a":1}');
Affected Rows: 1
SELECT host, ts, j, f FROM json2_alter_backfill WHERE j IS NOT NULL;
+------+-------------------------+---------+------+
| host | ts | j | f |
+------+-------------------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"a":1} | 10.0 |
+------+-------------------------+---------+------+
SELECT host, ts, j, f FROM json2_alter_backfill WHERE j.a = '1';
+------+-------------------------+---------+------+
| host | ts | j | f |
+------+-------------------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"a":1} | 10.0 |
+------+-------------------------+---------+------+
ADMIN FLUSH_TABLE('json2_alter_backfill');
+-------------------------------------------+
| ADMIN FLUSH_TABLE('json2_alter_backfill') |
+-------------------------------------------+
| 0 |
+-------------------------------------------+
SELECT host, ts, j, f FROM json2_alter_backfill WHERE j IS NOT NULL;
+------+-------------------------+---------+------+
| host | ts | j | f |
+------+-------------------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"a":1} | 10.0 |
+------+-------------------------+---------+------+
ADMIN COMPACT_TABLE('json2_alter_backfill');
+---------------------------------------------+
| ADMIN COMPACT_TABLE('json2_alter_backfill') |
+---------------------------------------------+
| 0 |
+---------------------------------------------+
-- SQLNESS ARG restart=true
SELECT host, ts, j, f FROM json2_alter_backfill WHERE j IS NOT NULL;
+------+-------------------------+---------+------+
| host | ts | j | f |
+------+-------------------------+---------+------+
| one | 1970-01-01T00:00:00.001 | {"a":1} | 10.0 |
+------+-------------------------+---------+------+
DROP TABLE json2_alter_backfill;
Affected Rows: 0
tests/cases/standalone/common/types/json/json2_non_append.sql
+201
View File
@@ -0,0 +1,201 @@
create table json2_without_append_mode (
ts timestamp time index,
j json2
);
drop table json2_without_append_mode;
create table json2_append_mode_false (
ts timestamp time index,
j json2
) with (
'append_mode' = 'false'
);
drop table json2_append_mode_false;
create table json2_alter_non_append (
ts timestamp time index
);
insert into json2_alter_non_append values (1);
alter table json2_alter_non_append add column j json2;
insert into json2_alter_non_append values (1, '{"a":1}');
select ts, j.a from json2_alter_non_append;
drop table json2_alter_non_append;
-- JSON2 is merged as a whole field, including empty objects and explicit JSON nulls.
CREATE TABLE json2_last_row (
host STRING,
ts TIMESTAMP TIME INDEX,
j JSON2,
k JSON2(max_auto_expanded_paths = 0),
f DOUBLE,
PRIMARY KEY (host)
) WITH (
'append_mode' = 'false',
'merge_mode' = 'last_row'
);
INSERT INTO json2_last_row VALUES
('one', 1, '{"a":1}', '{"a":1}', 10),
('one', 2, '{"a":2}', '{"a":2}', 10),
('one', 3, '{"a":3}', '{"a":3}', 10),
('one', 4, '{"a":4}', '{"a":4}', 10),
('one', 5, '{"a":{"x":5}}', '{"a":{"x":5}}', 10),
('one', 6, '{"a":{"x":6}}', '{"a":{"x":6}}', 10),
('one', 7, '{"a":{"x":8}}', '{"a":{"x":8}}', 10),
('two', 1, '{"a":99}', '{"a":99}', 99);
ADMIN FLUSH_TABLE('json2_last_row');
-- New objects replace old objects; only SQL NULL can fall back in last_non_null.
INSERT INTO json2_last_row VALUES
('one', 1, '{"b":2}', '{"b":2}', NULL),
('one', 2, '{}', '{}', NULL),
('one', 3, '{"a":null}', '{"a":null}', NULL),
('one', 4, NULL, NULL, NULL),
('one', 5, '{"a":42}', '{"a":42}', NULL),
('one', 6, '{"a":{"x":7}}', '{"a":{"x":7}}', NULL),
('one', 7, NULL, NULL, NULL);
-- Path-only reads must keep the new nested value and distinguish SQL NULL fallback.
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_row ORDER BY host, ts;
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts;
SELECT host, ts FROM json2_last_row WHERE j.a = '1' ORDER BY host, ts;
ADMIN FLUSH_TABLE('json2_last_row');
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_row ORDER BY host, ts;
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts;
ADMIN COMPACT_TABLE('json2_last_row');
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_row ORDER BY host, ts;
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts;
-- SQLNESS ARG restart=true
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_row ORDER BY host, ts;
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts;
-- A delete must prevent last_non_null from filling fields from older versions.
DELETE FROM json2_last_row WHERE host = 'one' AND ts = 1;
INSERT INTO json2_last_row VALUES ('one', 1, '{"c":3}', '{"c":3}', NULL);
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_row ORDER BY host, ts;
ADMIN FLUSH_TABLE('json2_last_row');
ADMIN COMPACT_TABLE('json2_last_row');
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_row ORDER BY host, ts;
DELETE FROM json2_last_row WHERE j.a = '99';
SELECT COUNT(*) FROM json2_last_row WHERE host = 'two';
DROP TABLE json2_last_row;
-- JSON2 is merged as a whole field, including empty objects and explicit JSON nulls.
CREATE TABLE json2_last_non_null (
host STRING,
ts TIMESTAMP TIME INDEX,
j JSON2,
k JSON2(max_auto_expanded_paths = 0),
f DOUBLE,
PRIMARY KEY (host)
) WITH (
'append_mode' = 'false',
'merge_mode' = 'last_non_null'
);
INSERT INTO json2_last_non_null VALUES
('one', 1, '{"a":1}', '{"a":1}', 10),
('one', 2, '{"a":2}', '{"a":2}', 10),
('one', 3, '{"a":3}', '{"a":3}', 10),
('one', 4, '{"a":4}', '{"a":4}', 10),
('one', 5, '{"a":{"x":5}}', '{"a":{"x":5}}', 10),
('one', 6, '{"a":{"x":6}}', '{"a":{"x":6}}', 10),
('one', 7, '{"a":{"x":8}}', '{"a":{"x":8}}', 10),
('two', 1, '{"a":99}', '{"a":99}', 99);
ADMIN FLUSH_TABLE('json2_last_non_null');
-- New objects replace old objects; only SQL NULL can fall back in last_non_null.
INSERT INTO json2_last_non_null VALUES
('one', 1, '{"b":2}', '{"b":2}', NULL),
('one', 2, '{}', '{}', NULL),
('one', 3, '{"a":null}', '{"a":null}', NULL),
('one', 4, NULL, NULL, NULL),
('one', 5, '{"a":42}', '{"a":42}', NULL),
('one', 6, '{"a":{"x":7}}', '{"a":{"x":7}}', NULL),
('one', 7, NULL, NULL, NULL);
-- Path-only reads must keep the new nested value and distinguish SQL NULL fallback.
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_non_null ORDER BY host, ts;
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts;
SELECT host, ts FROM json2_last_non_null WHERE j.a = '1' ORDER BY host, ts;
ADMIN FLUSH_TABLE('json2_last_non_null');
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_non_null ORDER BY host, ts;
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts;
ADMIN COMPACT_TABLE('json2_last_non_null');
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_non_null ORDER BY host, ts;
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts;
-- SQLNESS ARG restart=true
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_non_null ORDER BY host, ts;
SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts;
-- A delete must prevent last_non_null from filling fields from older versions.
DELETE FROM json2_last_non_null WHERE host = 'one' AND ts = 1;
INSERT INTO json2_last_non_null VALUES ('one', 1, '{"c":3}', '{"c":3}', NULL);
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_non_null ORDER BY host, ts;
ADMIN FLUSH_TABLE('json2_last_non_null');
ADMIN COMPACT_TABLE('json2_last_non_null');
SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f
FROM json2_last_non_null ORDER BY host, ts;
DELETE FROM json2_last_non_null WHERE j.a = '99';
SELECT COUNT(*) FROM json2_last_non_null WHERE host = 'two';
DROP TABLE json2_last_non_null;
-- A predicate on an added JSON2 field must retain older values of other fields.
CREATE TABLE json2_alter_backfill (
host STRING,
ts TIMESTAMP TIME INDEX,
f DOUBLE,
PRIMARY KEY (host)
) WITH ('merge_mode' = 'last_non_null');
INSERT INTO json2_alter_backfill VALUES ('one', 1, 10);
ADMIN FLUSH_TABLE('json2_alter_backfill');
ALTER TABLE json2_alter_backfill ADD COLUMN j JSON2;
INSERT INTO json2_alter_backfill VALUES ('one', 1, NULL, '{"a":1}');
SELECT host, ts, j, f FROM json2_alter_backfill WHERE j IS NOT NULL;
SELECT host, ts, j, f FROM json2_alter_backfill WHERE j.a = '1';
ADMIN FLUSH_TABLE('json2_alter_backfill');
SELECT host, ts, j, f FROM json2_alter_backfill WHERE j IS NOT NULL;
ADMIN COMPACT_TABLE('json2_alter_backfill');
-- SQLNESS ARG restart=true
SELECT host, ts, j, f FROM json2_alter_backfill WHERE j IS NOT NULL;
DROP TABLE json2_alter_backfill;