github-actions[bot] commented on code in PR #66421:
URL: https://github.com/apache/doris/pull/66421#discussion_r3710264289
##########
be/test/format_v2/parquet/variant_column_reader_test.cpp:
##########
@@ -776,6 +1109,273 @@ TEST(VariantColumnReaderTest,
MaterializesShreddedArrayElements) {
EXPECT_EQ(value.array_at(1).get_int(), 4);
}
+TEST(VariantColumnReaderTest, RejectsCorruptShreddedWrappersWithoutCrashing) {
+ const std::array<char, 2> int_seven {
+ static_cast<char>(static_cast<uint8_t>(VariantPrimitiveId::INT8)
+ << VARIANT_VALUE_HEADER_SHIFT),
+ 7};
+ const std::array<char, 1> invalid_value {static_cast<char>(0xff)};
+ const std::array<char, 1> ignored {0};
+ const StringRef metadata(VARIANT_EMPTY_METADATA.data(),
VARIANT_EMPTY_METADATA.size());
+ const StringRef residual_int(int_seven.data(), int_seven.size());
+ auto expect_error = [](const std::string& error, std::string_view
expected) {
+ EXPECT_NE(error.find(expected), std::string::npos) << error;
+ };
+ std::string_view current_case;
+
+ try {
+ {
+ current_case = "null metadata";
+ SCOPED_TRACE("null metadata");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {1}));
+ fields.push_back(nullable_strings({residual_int}, {0}));
+ expect_error(
+ materialization_error(unshredded_schema(),
root_wrapper(std::move(fields))),
+ "null metadata");
+ }
+ {
+ current_case = "wrapper without carriers";
+ SCOPED_TRACE("wrapper without carriers");
+ auto schema = unshredded_schema();
+ schema.children.pop_back();
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ expect_error(materialization_error(schema,
root_wrapper(std::move(fields))),
+ "neither value nor typed_value");
+ }
+ {
+ current_case = "scalar with residual";
+ SCOPED_TRACE("scalar with residual");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({residual_int}, {0}));
+ fields.push_back(nullable_int64({8}, {0}));
+ expect_error(
+ materialization_error(shredded_int64_schema(),
root_wrapper(std::move(fields))),
+ "scalar typed_value cannot have residual");
+ }
+ {
+ current_case = "object with scalar residual";
+ SCOPED_TRACE("object with scalar residual");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({residual_int}, {0}));
+ MutableColumns wrapper_fields;
+ wrapper_fields.push_back(nullable_int64({9}, {0}));
+ MutableColumns object_fields;
+ object_fields.push_back(ColumnNullable::create(
+ ColumnStruct::create(std::move(wrapper_fields)),
ColumnUInt8::create(1, 0)));
+
fields.push_back(ColumnNullable::create(ColumnStruct::create(std::move(object_fields)),
+ ColumnUInt8::create(1,
0)));
+ expect_error(materialization_error(shredded_object_schema(),
+
root_wrapper(std::move(fields))),
+ "non-object residual");
+ }
+ {
+ current_case = "object field count mismatch";
+ SCOPED_TRACE("object field count mismatch");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({{ignored.data(), 0}}, {1}));
+ MutableColumns unexpected_object_fields;
+ unexpected_object_fields.push_back(nullable_int64({1}, {0}));
+ unexpected_object_fields.push_back(nullable_int64({2}, {0}));
+ fields.push_back(ColumnNullable::create(
+ ColumnStruct::create(std::move(unexpected_object_fields)),
+ ColumnUInt8::create(1, 0)));
+ expect_error(materialization_error(shredded_object_schema(),
+
root_wrapper(std::move(fields))),
+ "physical field count mismatch");
+ }
+ {
+ current_case = "array with residual";
+ SCOPED_TRACE("array with residual");
+ MutableColumns empty_wrapper_fields;
+ empty_wrapper_fields.push_back(nullable_int64({}, {}));
+ auto empty_elements = ColumnNullable::create(
+ ColumnStruct::create(std::move(empty_wrapper_fields)),
ColumnUInt8::create());
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->insert_value(0);
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({residual_int}, {0}));
+ fields.push_back(ColumnNullable::create(
+ ColumnArray::create(std::move(empty_elements),
std::move(offsets)),
+ ColumnUInt8::create(1, 0)));
+ expect_error(
+ materialization_error(shredded_array_schema(),
root_wrapper(std::move(fields))),
+ "array typed_value cannot have residual");
+ }
+ {
+ current_case = "null array element wrapper";
+ SCOPED_TRACE("null array element wrapper");
+ MutableColumns wrapper_fields;
+ wrapper_fields.push_back(nullable_int64({0}, {1}));
+ auto wrappers = ColumnStruct::create(std::move(wrapper_fields));
+ auto elements = ColumnNullable::create(std::move(wrappers),
ColumnUInt8::create(1, 1));
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->insert_value(1);
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({{ignored.data(), 0}}, {1}));
+ fields.push_back(ColumnNullable::create(
+ ColumnArray::create(std::move(elements),
std::move(offsets)),
+ ColumnUInt8::create(1, 0)));
+ expect_error(
+ materialization_error(shredded_array_schema(),
root_wrapper(std::move(fields))),
+ "array element wrapper is null");
+ }
+ {
+ current_case = "missing array element";
+ SCOPED_TRACE("missing array element");
+ MutableColumns element_fields;
+ element_fields.push_back(nullable_strings({{ignored.data(), 0}},
{1}));
+ element_fields.push_back(nullable_int64({0}, {1}));
+ auto elements =
ColumnNullable::create(ColumnStruct::create(std::move(element_fields)),
+ ColumnUInt8::create(1, 0));
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->insert_value(1);
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({{ignored.data(), 0}}, {1}));
+ fields.push_back(ColumnNullable::create(
+ ColumnArray::create(std::move(elements),
std::move(offsets)),
+ ColumnUInt8::create(1, 0)));
+ expect_error(materialization_error(shredded_mixed_array_schema(),
+
root_wrapper(std::move(fields))),
+ "array element is missing");
+ }
+ {
+ current_case = "root field count mismatch";
+ SCOPED_TRACE("root field count mismatch");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({residual_int}, {0}));
+ fields.push_back(nullable_int64({8}, {0}));
+ fields.push_back(nullable_int64({9}, {0}));
+ expect_error(
+ materialization_error(shredded_int64_schema(),
root_wrapper(std::move(fields))),
+ "physical field count mismatch");
+ }
+ {
+ current_case = "invalid metadata";
+ SCOPED_TRACE("invalid metadata");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({StringRef("bad")}, {0}));
+ fields.push_back(nullable_strings({residual_int}, {0}));
+ expect_error(
+ materialization_error(unshredded_schema(),
root_wrapper(std::move(fields))),
+ "metadata");
+ }
+ {
+ current_case = "invalid residual value";
+ SCOPED_TRACE("invalid residual value");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({{invalid_value.data(),
invalid_value.size()}}, {0}));
+ expect_error(
+ materialization_error(unshredded_schema(),
root_wrapper(std::move(fields))),
+ "Variant");
+ }
+ } catch (const std::exception& error) {
+ FAIL() << "Unexpected exception in " << current_case << ": " <<
error.what();
+ }
+}
+
+TEST(VariantColumnReaderTest, ImmediateCorruptionLeavesDestinationUnchanged) {
+ auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
+ ASSERT_TRUE(
+ materialize_variant_rows(shredded_int64_schema(),
shredded_int64_physical({7}), output)
+ .ok());
+ MutableColumns invalid_fields;
+ invalid_fields.push_back(nullable_strings(
+ {{VARIANT_EMPTY_METADATA.data(), VARIANT_EMPTY_METADATA.size()}},
{0}));
+ const Status status = materialize_variant_rows(shredded_int64_schema(),
Review Comment:
[P1] Exercise the lazy append failure in this atomicity test
This malformed batch has only `metadata`, so `ParquetVariantShreddedState`
rejects its field count before `append_materialized_column()` mutates anything.
It therefore cannot catch the dangerous nested case: compatibility validation
accepts a Variant child without materializing it, STRUCT append mutates earlier
siblings in order, and a later incompatible/corrupt shredded Variant can throw
during `insert_range_from`, leaving those siblings grown despite the error
Status. Please add a pre-populated STRUCT/LIST/MAP case with a primitive
sibling before the Variant and corruption that is discovered only during lazy
fallback, then assert every child/null-map/offset remains unchanged (and fix
the transactional append path if it fails).
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy:
##########
@@ -337,44 +586,167 @@ suite("test_iceberg_variant_read",
ORDER BY id
"""
- // Keep the root Variant as output while the implicit scalar comparison
drives the shredded
- // typed_value statistics/page-index path.
- order_qt_variant_implicit_shredded_filter """
+ order_qt_variant_multi_file_serial """
+ SELECT id,
+ CAST(v['shared'] AS INT),
+ CAST(v['a'] AS INT),
+ CAST(v['b'] AS INT),
+ CAST(v['new_field']['k'] AS INT),
+ CAST(v AS STRING)
+ FROM variant_multi_file
+ WHERE v['shared'] >= 20
+ ORDER BY id
+ """
+ sql "set parallel_pipeline_task_num=4"
Review Comment:
[P2] Prove that the parallel case actually uses multiple scanners
These settings only raise concurrency ceilings;
`min_file_scanners_concurrency` remains 1, so the query may run serially and
still match the byte-for-byte serial result above. That leaves cross-file state
races uncovered. Force a minimum concurrency greater than one, assert the
fixture has multiple data files, and bind this query to a completed profile
showing actual concurrent/non-empty scanners rather than only a configured
maximum.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy:
##########
@@ -219,6 +437,25 @@ suite("test_iceberg_variant_read",
}
return sum
}
+ def getProfileByToken = { String token, List<String> positiveCounters = []
->
+ String lastProfile = ""
+ for (int retry = 0; retry < 20; ++retry) {
+ List profileData = profileAction.getProfileList()
Review Comment:
[P1] Use the completion-aware profile waiter
This custom loop gives asynchronously published scanner counters only 10
seconds and never requires the profile list/detail to be COMPLETE.
`ProfileAction.getProfileBySql` already provides a completion-aware 60-second
wait, and the existing DV suite explicitly notes that detailed scanner counters
arrive after query return. On a loaded external CI run these new multi-counter
calls can fail even though the query and final profile are correct. Please use
the framework waiter (then poll positivity if needed) without shortening its
established timeout.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy:
##########
@@ -100,6 +133,115 @@ suite("test_iceberg_variant_read",
(10,
parse_json('{"name":"dave","n":50,"ratio":5.5,"ok":false,"arr":[7,8],"nested":{"city":"sz"}}')),
(11,
parse_json('{"name":null,"n":60,"ratio":6.5,"ok":true,"arr":[9,10],"nested":{"city":null}}'));
+ DROP TABLE IF EXISTS demo.${dbName}.variant_root_arrays;
+ CREATE TABLE demo.${dbName}.variant_root_arrays (id INT, v VARIANT)
USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='true',
+ 'write.parquet.variant-inference-buffer-size'='100'
+ );
+ INSERT INTO demo.${dbName}.variant_root_arrays VALUES
+ (1, parse_json('[]')),
+ (2, parse_json('[null,1,{"x":2},[3,4],"tail"]')),
+ (3, parse_json('[{"nested":[null,{"y":5}]}]')),
+ (4, parse_json('null')),
+ (5, NULL);
+
+ DROP TABLE IF EXISTS demo.${dbName}.variant_multi_file;
+ CREATE TABLE demo.${dbName}.variant_multi_file (id INT, v VARIANT)
USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='false'
+ );
+ INSERT INTO demo.${dbName}.variant_multi_file
+ VALUES (1, parse_json('{"a":1,"shared":10}'));
+ ALTER TABLE demo.${dbName}.variant_multi_file SET TBLPROPERTIES (
+ 'write.parquet.shred-variants'='true',
+ 'write.parquet.variant-inference-buffer-size'='1'
+ );
+ INSERT INTO demo.${dbName}.variant_multi_file
+ VALUES (2, parse_json('{"b":2,"shared":20,"z":200}'));
+ INSERT INTO demo.${dbName}.variant_multi_file
+ VALUES (3, parse_json('{"z":300,"shared":30,"a":3}'));
+ ALTER TABLE demo.${dbName}.variant_multi_file SET TBLPROPERTIES
+ ('write.parquet.shred-variants'='false');
+ INSERT INTO demo.${dbName}.variant_multi_file
+ VALUES (4, parse_json('{"c":4,"shared":40}'));
+ ALTER TABLE demo.${dbName}.variant_multi_file SET TBLPROPERTIES
+ ('write.parquet.shred-variants'='true');
+ INSERT INTO demo.${dbName}.variant_multi_file
+ VALUES (5,
parse_json('{"shared":50,"b":5,"new_field":{"k":500}}'));
+
+ DROP TABLE IF EXISTS demo.${dbName}.variant_type_matrix;
+ CREATE TABLE demo.${dbName}.variant_type_matrix (id INT, v VARIANT)
USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='true',
+ 'write.parquet.variant-inference-buffer-size'='100'
+ );
+ INSERT INTO demo.${dbName}.variant_type_matrix SELECT 1,
to_variant_object(named_struct(
+ 'bool_value', true,
+ 'tiny_value', CAST(-128 AS TINYINT),
+ 'small_value', CAST(-32768 AS SMALLINT),
+ 'int_value', CAST(2147483647 AS INT),
+ 'big_value', CAST('-9223372036854775808' AS BIGINT),
+ 'float_value', CAST('NaN' AS FLOAT),
+ 'double_value', CAST('Infinity' AS DOUBLE),
+ 'decimal_value', CAST('-1234567890.1234' AS DECIMAL(20, 4)),
+ 'date_value', CAST('1970-01-02' AS DATE),
+ 'timestamp_value', TIMESTAMP'1970-01-01 00:00:01.234567',
+ 'binary_value', CAST('binary' AS BINARY),
+ 'null_value', CAST(NULL AS INT)
+ ));
+
+ DROP TABLE IF EXISTS demo.${dbName}.variant_multi_row_group;
+ CREATE TABLE demo.${dbName}.variant_multi_row_group (id INT, v
VARIANT) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='true',
Review Comment:
[P1] Make this fixture's physical Variant mode explicit
This table requests `write.parquet.shred-variants=true`, but the checks
below require `VariantDirectLeafPathMisses` and reconstruction and call the
file unshredded. In the repository's pinned Iceberg 1.10.1 Spark runtime the
property is not defined in
[TableProperties](https://github.com/apache/iceberg/blob/apache-iceberg-1.10.1/core/src/main/java/org/apache/iceberg/TableProperties.java),
so the test passes only because it is ignored; a writer that honors it will
shred `n` and make the fallback profile time out. The other new Spark-generated
`true` fixtures likewise do not prove shredded decoding under the current pin.
Set `false` for fallback fixtures, and use a capable/checked physical fixture
plus a schema or direct-leaf oracle for cases intended to cover shredding.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy:
##########
@@ -337,44 +586,167 @@ suite("test_iceberg_variant_read",
ORDER BY id
"""
- // Keep the root Variant as output while the implicit scalar comparison
drives the shredded
- // typed_value statistics/page-index path.
- order_qt_variant_implicit_shredded_filter """
+ order_qt_variant_multi_file_serial """
+ SELECT id,
+ CAST(v['shared'] AS INT),
+ CAST(v['a'] AS INT),
+ CAST(v['b'] AS INT),
+ CAST(v['new_field']['k'] AS INT),
+ CAST(v AS STRING)
+ FROM variant_multi_file
+ WHERE v['shared'] >= 20
+ ORDER BY id
+ """
+ sql "set parallel_pipeline_task_num=4"
+ sql "set max_file_scanners_concurrency=8"
+ order_qt_variant_multi_file_parallel """
+ SELECT id,
+ CAST(v['shared'] AS INT),
+ CAST(v['a'] AS INT),
+ CAST(v['b'] AS INT),
+ CAST(v['new_field']['k'] AS INT),
+ CAST(v AS STRING)
+ FROM variant_multi_file
+ WHERE v['shared'] >= 20
+ ORDER BY id
+ """
+
+ order_qt_variant_type_matrix """
+ SELECT CAST(v['bool_value'] AS BOOLEAN),
+ CAST(v['tiny_value'] AS TINYINT),
+ CAST(v['small_value'] AS SMALLINT),
+ CAST(v['int_value'] AS INT),
+ CAST(v['big_value'] AS BIGINT),
+ ISNAN(CAST(v['float_value'] AS FLOAT)),
+ ISINF(CAST(v['double_value'] AS DOUBLE)),
+ CAST(v['decimal_value'] AS DECIMAL(20, 4)),
+ CAST(v['date_value'] AS DATE),
+ CAST(v['timestamp_value'] AS DATETIMEV2(6)),
+ CAST(v['binary_value'] AS STRING),
+ v['null_value'] IS NULL
+ FROM variant_type_matrix
+ """
+
+ String multiRowGroupColdToken =
+ "iceberg_variant_multi_row_group_cold_" +
UUID.randomUUID().toString()
+ sql """
+ SELECT '${multiRowGroupColdToken}', COUNT(*), MIN(id), MAX(id)
+ FROM variant_multi_row_group
+ WHERE CAST(v['n'] AS INT) >= 8000
+ """
+ String multiRowGroupColdProfile = getProfileByToken(multiRowGroupColdToken,
+ ["RowGroupsTotalNum", "VariantDirectLeafPathMisses",
"VariantReconstructedRows",
+ "FilteredRowsByLazyRead"]).toString()
+ assertTrue(counterSum(multiRowGroupColdProfile, "RowGroupsTotalNum") > 1,
+ "The generated Variant file did not contain multiple Parquet
row groups")
+ assertTrue(counterSum(multiRowGroupColdProfile,
"VariantDirectLeafPathMisses") > 0,
+ "The unshredded scan did not record its direct-leaf fallback")
+ assertTrue(counterSum(multiRowGroupColdProfile,
"VariantReconstructedRows") > 0,
+ "The unshredded scan did not reconstruct Variant rows")
+ assertTrue(counterSum(multiRowGroupColdProfile, "FilteredRowsByLazyRead")
> 0,
+ "The unshredded Variant predicate did not defer non-predicate
columns")
+ String multiRowGroupWarmToken =
+ "iceberg_variant_multi_row_group_warm_" +
UUID.randomUUID().toString()
+ sql """
+ SELECT '${multiRowGroupWarmToken}', COUNT(*), MIN(id), MAX(id)
+ FROM variant_multi_row_group
+ WHERE CAST(v['n'] AS INT) >= 8000
+ """
+ String multiRowGroupWarmProfile = getProfileByToken(multiRowGroupWarmToken,
+ ["VariantDirectLeafPathMisses"]).toString()
+ assertTrue(counterSum(multiRowGroupWarmProfile,
"VariantDirectLeafPathMisses") > 0,
+ "The warm unshredded scan did not preserve its direct-leaf
fallback")
+ qt_variant_multi_row_group_result """
+ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT))
+ FROM variant_multi_row_group
+ WHERE CAST(v['n'] AS INT) >= 8000
+ """
+
+ qt_variant_deletion_vector_current """
+ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT))
+ FROM variant_deletion_vector
+ WHERE v['keep'] = true
+ """
+ qt_variant_deletion_vector_before_delete """
+ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT))
+ FROM variant_deletion_vector FOR VERSION AS OF
${deletionVectorBaseSnapshot}
+ WHERE v['n'] >= 0
+ """
+ order_qt_variant_equality_delete_current """
+ SELECT id, CAST(v['n'] AS INT), CAST(v['label'] AS STRING), CAST(v AS
STRING)
+ FROM variant_equality_delete
+ WHERE v['n'] >= 0
+ ORDER BY id
+ """
+ order_qt_variant_equality_delete_before_delete """
+ SELECT id, CAST(v['n'] AS INT), CAST(v['label'] AS STRING), CAST(v AS
STRING)
+ FROM variant_equality_delete FOR VERSION AS OF
${equalityDeleteBaseSnapshot}
+ WHERE v['n'] >= 0
+ ORDER BY id
+ """
+
+ // Keep the root Variant as output while the scalar comparison exercises
the fallback path for
+ // the unshredded Spark files.
+ order_qt_variant_implicit_filter """
SELECT id, CAST(v AS STRING)
FROM variant_values
WHERE v['n'] > 35
ORDER BY id
"""
- // The query projects the root Variant, while the predicate uses
typed_value page metadata.
+ qt_variant_shredded_only_time_travel """
+ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT))
+ FROM variant_page_pruning FOR VERSION AS OF ${shreddedOnlySnapshot}
+ WHERE CAST(v['n'] AS INT) > 3000
+ """
+ qt_variant_mixed_before_delete """
+ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT))
+ FROM variant_page_pruning FOR VERSION AS OF
${mixedBeforeDeleteSnapshot}
+ WHERE CAST(v['n'] AS INT) > 3000
+ """
+
+ // The query projects the complete Variant while its predicate reads the
shredded typed leaf.
+ // The appended unshredded file must fall back independently in the same
scan.
String pagePruningToken = "iceberg_variant_page_pruning_" +
UUID.randomUUID().toString()
sql """
SELECT '${pagePruningToken}', id, CAST(v AS STRING)
FROM variant_page_pruning
- WHERE v['n'] > 3000
+ WHERE CAST(v['n'] AS INT) > 3000
ORDER BY id
"""
- String pagePruningProfile = getProfileByToken(pagePruningToken).toString()
+ String pagePruningProfile = getProfileByToken(pagePruningToken,
+ ["FilteredRowsByPage", "VariantLeafProjections",
"VariantDirectLeafPathMisses",
+ "VariantDirectLeafRows", "VariantReconstructedRows",
+ "FilteredRowsByLazyRead"]).toString()
assertTrue(counterSum(pagePruningProfile, "FilteredRowsByPage") > 0,
"Shredded Variant typed_value did not filter any Parquet page")
// The predicate_access_paths contract keeps the typed leaf eager while
the complete Variant
// root is read through the independent deferred-output projection.
assertTrue(counterSum(pagePruningProfile, "VariantLeafProjections") > 0,
"A root Variant output query did not retain its typed predicate
leaf projection")
+ assertTrue(counterSum(pagePruningProfile, "VariantDirectLeafPathMisses") >
0,
+ "The mixed scan did not fall back for its unshredded Variant
file")
+ assertTrue(counterSum(pagePruningProfile, "VariantDirectLeafRows") > 0,
+ "The mixed scan did not evaluate rows from the shredded typed
leaf")
+ assertTrue(counterSum(pagePruningProfile, "VariantReconstructedRows") > 0,
+ "The mixed scan did not reconstruct complete Variant output")
+ assertTrue(counterSum(pagePruningProfile, "FilteredRowsByLazyRead") > 0,
Review Comment:
[P1] Make delayed materialization specific to the Variant output
`FilteredRowsByLazyRead` is incremented whenever any non-predicate column is
deferred. Because this query also outputs `id`, the counter stays positive if
only `id` is lazy while the complete Variant root regresses to eager
reconstruction; `VariantReconstructedRows > 0` does not distinguish that case.
Assert a Variant-specific relationship (for example reconstructed rows equal
selected/output rows and are below pre-filter candidates), or add/use a
Variant-specific deferred-materialization counter.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy:
##########
@@ -337,44 +586,167 @@ suite("test_iceberg_variant_read",
ORDER BY id
"""
- // Keep the root Variant as output while the implicit scalar comparison
drives the shredded
- // typed_value statistics/page-index path.
- order_qt_variant_implicit_shredded_filter """
+ order_qt_variant_multi_file_serial """
+ SELECT id,
+ CAST(v['shared'] AS INT),
+ CAST(v['a'] AS INT),
+ CAST(v['b'] AS INT),
+ CAST(v['new_field']['k'] AS INT),
+ CAST(v AS STRING)
+ FROM variant_multi_file
+ WHERE v['shared'] >= 20
+ ORDER BY id
+ """
+ sql "set parallel_pipeline_task_num=4"
+ sql "set max_file_scanners_concurrency=8"
+ order_qt_variant_multi_file_parallel """
+ SELECT id,
+ CAST(v['shared'] AS INT),
+ CAST(v['a'] AS INT),
+ CAST(v['b'] AS INT),
+ CAST(v['new_field']['k'] AS INT),
+ CAST(v AS STRING)
+ FROM variant_multi_file
+ WHERE v['shared'] >= 20
+ ORDER BY id
+ """
+
+ order_qt_variant_type_matrix """
+ SELECT CAST(v['bool_value'] AS BOOLEAN),
+ CAST(v['tiny_value'] AS TINYINT),
+ CAST(v['small_value'] AS SMALLINT),
+ CAST(v['int_value'] AS INT),
+ CAST(v['big_value'] AS BIGINT),
+ ISNAN(CAST(v['float_value'] AS FLOAT)),
+ ISINF(CAST(v['double_value'] AS DOUBLE)),
+ CAST(v['decimal_value'] AS DECIMAL(20, 4)),
+ CAST(v['date_value'] AS DATE),
+ CAST(v['timestamp_value'] AS DATETIMEV2(6)),
+ CAST(v['binary_value'] AS STRING),
+ v['null_value'] IS NULL
+ FROM variant_type_matrix
+ """
+
+ String multiRowGroupColdToken =
+ "iceberg_variant_multi_row_group_cold_" +
UUID.randomUUID().toString()
+ sql """
+ SELECT '${multiRowGroupColdToken}', COUNT(*), MIN(id), MAX(id)
+ FROM variant_multi_row_group
+ WHERE CAST(v['n'] AS INT) >= 8000
+ """
+ String multiRowGroupColdProfile = getProfileByToken(multiRowGroupColdToken,
+ ["RowGroupsTotalNum", "VariantDirectLeafPathMisses",
"VariantReconstructedRows",
+ "FilteredRowsByLazyRead"]).toString()
+ assertTrue(counterSum(multiRowGroupColdProfile, "RowGroupsTotalNum") > 1,
+ "The generated Variant file did not contain multiple Parquet
row groups")
+ assertTrue(counterSum(multiRowGroupColdProfile,
"VariantDirectLeafPathMisses") > 0,
+ "The unshredded scan did not record its direct-leaf fallback")
+ assertTrue(counterSum(multiRowGroupColdProfile,
"VariantReconstructedRows") > 0,
+ "The unshredded scan did not reconstruct Variant rows")
+ assertTrue(counterSum(multiRowGroupColdProfile, "FilteredRowsByLazyRead")
> 0,
+ "The unshredded Variant predicate did not defer non-predicate
columns")
+ String multiRowGroupWarmToken =
+ "iceberg_variant_multi_row_group_warm_" +
UUID.randomUUID().toString()
+ sql """
+ SELECT '${multiRowGroupWarmToken}', COUNT(*), MIN(id), MAX(id)
+ FROM variant_multi_row_group
+ WHERE CAST(v['n'] AS INT) >= 8000
+ """
+ String multiRowGroupWarmProfile = getProfileByToken(multiRowGroupWarmToken,
+ ["VariantDirectLeafPathMisses"]).toString()
+ assertTrue(counterSum(multiRowGroupWarmProfile,
"VariantDirectLeafPathMisses") > 0,
+ "The warm unshredded scan did not preserve its direct-leaf
fallback")
+ qt_variant_multi_row_group_result """
+ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT))
+ FROM variant_multi_row_group
+ WHERE CAST(v['n'] AS INT) >= 8000
+ """
+
+ qt_variant_deletion_vector_current """
+ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT))
+ FROM variant_deletion_vector
+ WHERE v['keep'] = true
Review Comment:
[P1] Make the DV checkpoint observe the deleted rows
The DELETE removes odd ids, and those rows were initialized with
`keep=false`; this `WHERE v['keep'] = true` therefore returns the same 2048
even rows and sum even if Doris ignores the delete entirely. It also cannot
distinguish a PUFFIN deletion vector from an ordinary position-delete file.
Query a predicate that includes the deleted rows (for example `v['n'] >= 0`)
and assert the current even-id result, then verify the live delete metadata is
PUFFIN/content-offset DV data for this fixture.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy:
##########
@@ -337,44 +586,167 @@ suite("test_iceberg_variant_read",
ORDER BY id
"""
- // Keep the root Variant as output while the implicit scalar comparison
drives the shredded
- // typed_value statistics/page-index path.
- order_qt_variant_implicit_shredded_filter """
+ order_qt_variant_multi_file_serial """
+ SELECT id,
+ CAST(v['shared'] AS INT),
+ CAST(v['a'] AS INT),
+ CAST(v['b'] AS INT),
+ CAST(v['new_field']['k'] AS INT),
+ CAST(v AS STRING)
+ FROM variant_multi_file
+ WHERE v['shared'] >= 20
+ ORDER BY id
+ """
+ sql "set parallel_pipeline_task_num=4"
+ sql "set max_file_scanners_concurrency=8"
+ order_qt_variant_multi_file_parallel """
+ SELECT id,
+ CAST(v['shared'] AS INT),
+ CAST(v['a'] AS INT),
+ CAST(v['b'] AS INT),
+ CAST(v['new_field']['k'] AS INT),
+ CAST(v AS STRING)
+ FROM variant_multi_file
+ WHERE v['shared'] >= 20
+ ORDER BY id
+ """
+
+ order_qt_variant_type_matrix """
+ SELECT CAST(v['bool_value'] AS BOOLEAN),
+ CAST(v['tiny_value'] AS TINYINT),
+ CAST(v['small_value'] AS SMALLINT),
+ CAST(v['int_value'] AS INT),
+ CAST(v['big_value'] AS BIGINT),
+ ISNAN(CAST(v['float_value'] AS FLOAT)),
+ ISINF(CAST(v['double_value'] AS DOUBLE)),
+ CAST(v['decimal_value'] AS DECIMAL(20, 4)),
+ CAST(v['date_value'] AS DATE),
+ CAST(v['timestamp_value'] AS DATETIMEV2(6)),
+ CAST(v['binary_value'] AS STRING),
+ v['null_value'] IS NULL
+ FROM variant_type_matrix
+ """
+
+ String multiRowGroupColdToken =
+ "iceberg_variant_multi_row_group_cold_" +
UUID.randomUUID().toString()
+ sql """
+ SELECT '${multiRowGroupColdToken}', COUNT(*), MIN(id), MAX(id)
+ FROM variant_multi_row_group
+ WHERE CAST(v['n'] AS INT) >= 8000
+ """
+ String multiRowGroupColdProfile = getProfileByToken(multiRowGroupColdToken,
+ ["RowGroupsTotalNum", "VariantDirectLeafPathMisses",
"VariantReconstructedRows",
+ "FilteredRowsByLazyRead"]).toString()
+ assertTrue(counterSum(multiRowGroupColdProfile, "RowGroupsTotalNum") > 1,
Review Comment:
[P1] Do not double-count merged and detailed profile views
`counterSum` scans the complete REST profile, which presents scanner
counters in both merged and detailed/execution sections. A single real
`RowGroupsTotalNum: 1` can therefore appear twice, sum to 2, and satisfy this
`> 1` assertion even though the file exercised only one row group. The existing
DV suite avoids this by slicing the authoritative `MergedProfile` section
before aggregation. Scope this threshold to one profile view (or parse one
ParquetReader block) and assert its actual row-group count.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
