hubgeter commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3648007631
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -532,33 +537,89 @@ public void createScanRangeLocations() throws
UserException {
super.createScanRangeLocations();
// Extract name mapping from Iceberg table properties
Map<Integer, List<String>> nameMapping = extractNameMapping();
+ Schema scanSchema = getQuerySchema();
+ List<Column> scanColumns = getScanColumns(scanSchema);
Review Comment:
Addressed on the current head. FE now preflights live equality-delete field
IDs under the catalog authenticator and adds only the referenced fields from
schema history to the recursive carrier. V1 and V2 tests cover add-default,
equality delete, and dropped-key materialization.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -532,33 +537,89 @@ public void createScanRangeLocations() throws
UserException {
super.createScanRangeLocations();
// Extract name mapping from Iceberg table properties
Map<Integer, List<String>> nameMapping = extractNameMapping();
+ Schema scanSchema = getQuerySchema();
+ List<Column> scanColumns = getScanColumns(scanSchema);
// Equality-delete keys are hidden scan dependencies and need not
appear in the query
// projection. Both scanners need the complete current schema to
resolve field ids,
// historical names, types, and initial defaults when an old data file
lacks such a key.
- ExternalUtil.initSchemaInfoForAllColumn(params, -1L,
source.getTargetTable().getColumns(),
- nameMapping, getBase64EncodedInitialDefaultsForScan());
+ ExternalUtil.initSchemaInfoForAllColumn(params, -1L, scanColumns,
nameMapping,
Review Comment:
Addressed on the current head. Recursive initial-default semantics use
iceberg_scan_semantics_version 2, and FE rejects smooth-upgrade source backends
only when the pruned projection requires the new recursive materialization.
Planner/unit coverage includes the mixed-version gate.
##########
be/test/format/table/iceberg/iceberg_reader_test.cpp:
##########
@@ -1049,6 +1270,341 @@ TEST_F(IcebergReaderTest,
v1_position_delete_read_error_releases_cache_entry) {
EXPECT_NE(status.to_string().find(delete_file.path), std::string::npos);
}
+TEST_F(IcebergReaderTest,
v1_rejects_missing_required_field_without_initial_default) {
+ schema::external::TField field;
+ field.__set_name("required_added");
+ field.__set_is_optional(false);
+
+ ColumnPtr column;
+ const auto status = iceberg::create_initial_default_column(
+ field, std::make_shared<DataTypeInt32>(), &column);
+
+ ASSERT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("required_added"), std::string::npos);
+ EXPECT_NE(status.to_string().find("no initial default"),
std::string::npos);
+}
+
+// GTest assertion macros inflate clang-tidy's cognitive-complexity score.
+// NOLINTNEXTLINE(readability-function-cognitive-complexity)
+TEST_F(IcebergReaderTest,
v1_reuses_prepared_complex_initial_default_across_block_types) {
+ schema::external::TField field;
+ field.__set_id(1);
+ field.__set_name("nested");
+ field.__set_is_optional(true);
+ field.__set_initial_default_value("{}");
+ auto child = std::make_shared<schema::external::TField>();
+ child->__set_id(2);
+ child->__set_name("added");
+ child->__set_is_optional(false);
+ child->__set_initial_default_value("7");
+ schema::external::TFieldPtr child_ptr;
+ child_ptr.__set_field_ptr(std::move(child));
+ schema::external::TStructField struct_field;
+ struct_field.__set_fields({child_ptr});
+ field.nestedField.__set_struct_field(struct_field);
+ field.__isset.nestedField = true;
+
+ auto make_type = [] {
+ return make_nullable(std::make_shared<DataTypeStruct>(
+ DataTypes {std::make_shared<DataTypeInt32>()}, Strings
{"added"}));
+ };
+ const auto first_type = make_type();
+ const auto second_type = make_type();
+ ASSERT_NE(first_type.get(), second_type.get());
+ ASSERT_TRUE(first_type->equals(*second_type));
+
+ std::unordered_map<int32_t, std::pair<DataTypePtr, ColumnPtr>>
prepared_values;
+ ColumnPtr first_batch = first_type->create_column();
+ ASSERT_TRUE(
+ iceberg::append_initial_default(field, first_type, 2,
&prepared_values, &first_batch)
+ .ok());
+ ASSERT_EQ(prepared_values.size(), 1);
+ const auto* prepared_value = prepared_values.at(field.id).second.get();
+
+ ColumnPtr second_batch = second_type->create_column();
+ ASSERT_TRUE(
+ iceberg::append_initial_default(field, second_type, 3,
&prepared_values, &second_batch)
+ .ok());
+ ASSERT_EQ(prepared_values.at(field.id).second.get(), prepared_value);
+ ASSERT_EQ(prepared_values.at(field.id).first.get(), first_type.get());
+ ASSERT_EQ(first_batch->size(), 2);
+ ASSERT_EQ(second_batch->size(), 3);
+ for (size_t row = 0; row < first_batch->size(); ++row) {
+ Field value;
+ first_batch->get(row, value);
+ ASSERT_EQ(value.get<TYPE_STRUCT>().size(), 1);
+ EXPECT_EQ(value.get<TYPE_STRUCT>()[0].get<TYPE_INT>(), 7);
+ }
+ for (size_t row = 0; row < second_batch->size(); ++row) {
+ Field value;
+ second_batch->get(row, value);
+ ASSERT_EQ(value.get<TYPE_STRUCT>().size(), 1);
+ EXPECT_EQ(value.get<TYPE_STRUCT>()[0].get<TYPE_INT>(), 7);
+ }
+}
+
+// GTest assertion macros inflate clang-tidy's cognitive-complexity score.
+// NOLINTNEXTLINE(readability-function-cognitive-complexity)
+TEST_F(IcebergReaderTest, v1_materializes_complex_initial_defaults) {
+ auto make_field = [](const std::string& name, int32_t id, bool optional,
+ std::optional<std::string> initial_default =
std::nullopt,
+ bool binary_like = false) {
+ auto field = std::make_shared<schema::external::TField>();
+ field->__set_name(name);
+ field->__set_id(id);
+ field->__set_is_optional(optional);
+ if (initial_default.has_value()) {
+ field->__set_initial_default_value(*initial_default);
+ }
+ if (binary_like) {
+ field->__set_initial_default_value_is_base64(true);
+ }
+ schema::external::TFieldPtr result;
+ result.__set_field_ptr(std::move(field));
+ return result;
+ };
+
+ const auto required_int_type = std::make_shared<DataTypeInt32>();
+ const auto optional_string_type =
make_nullable(std::make_shared<DataTypeString>());
+ const auto struct_type = make_nullable(
+ std::make_shared<DataTypeStruct>(DataTypes {required_int_type,
optional_string_type},
+ Strings {"required_added",
"optional_added"}));
+ auto required_added = make_field("required_added", 2, false, "7");
+ required_added.field_ptr->__set_name_mapping({"legacy_required_added"});
+ auto optional_added = make_field("optional_added", 3, true);
+ schema::external::TStructField struct_metadata;
+ struct_metadata.__set_fields({required_added, optional_added});
+ auto struct_field = make_field("struct_default", 1, true, "{}");
+ struct_field.field_ptr->nestedField.__set_struct_field(struct_metadata);
+ struct_field.field_ptr->__isset.nestedField = true;
+
+ ColumnPtr struct_column;
+
ASSERT_TRUE(iceberg::create_initial_default_column(*struct_field.field_ptr,
struct_type,
+ &struct_column)
+ .ok());
+ Field struct_value;
+ struct_column->get(0, struct_value);
+ const auto& struct_fields = struct_value.get<TYPE_STRUCT>();
+ ASSERT_EQ(struct_fields.size(), 2);
+ EXPECT_EQ(struct_fields[0].get<TYPE_INT>(), 7);
+ EXPECT_TRUE(struct_fields[1].is_null());
+
+ const auto pruned_struct_type =
make_nullable(std::make_shared<DataTypeStruct>(
+ DataTypes {required_int_type}, Strings {"legacy_required_added"}));
+ ColumnPtr pruned_struct_column;
+
ASSERT_TRUE(iceberg::create_initial_default_column(*struct_field.field_ptr,
pruned_struct_type,
+ &pruned_struct_column)
+ .ok());
+ Field pruned_struct_value;
+ pruned_struct_column->get(0, pruned_struct_value);
+ const auto& pruned_struct_fields = pruned_struct_value.get<TYPE_STRUCT>();
+ ASSERT_EQ(pruned_struct_fields.size(), 1);
+ EXPECT_EQ(pruned_struct_fields[0].get<TYPE_INT>(), 7);
+
+ const auto reordered_struct_type = make_nullable(
+ std::make_shared<DataTypeStruct>(DataTypes {optional_string_type,
required_int_type},
+ Strings {"optional_added",
"legacy_required_added"}));
+ ColumnPtr reordered_struct_column;
+ ASSERT_TRUE(iceberg::create_initial_default_column(
+ *struct_field.field_ptr, reordered_struct_type,
&reordered_struct_column)
+ .ok());
+ Field reordered_struct_value;
+ reordered_struct_column->get(0, reordered_struct_value);
+ const auto& reordered_struct_fields =
reordered_struct_value.get<TYPE_STRUCT>();
+ ASSERT_EQ(reordered_struct_fields.size(), 2);
+ EXPECT_TRUE(reordered_struct_fields[0].is_null());
+ EXPECT_EQ(reordered_struct_fields[1].get<TYPE_INT>(), 7);
+
+ const auto list_type =
make_nullable(std::make_shared<DataTypeArray>(optional_string_type));
+ auto list_element = make_field("element", 5, true);
+ schema::external::TArrayField list_metadata;
+ list_metadata.__set_item_field(list_element);
+ auto list_field = make_field("list_default", 4, true, "[\"alpha\",null]");
+ list_field.field_ptr->nestedField.__set_array_field(list_metadata);
+ list_field.field_ptr->__isset.nestedField = true;
+
+ ColumnPtr list_column;
+ ASSERT_TRUE(
+ iceberg::create_initial_default_column(*list_field.field_ptr,
list_type, &list_column)
+ .ok());
+ Field list_value;
+ list_column->get(0, list_value);
+ const auto& list_fields = list_value.get<TYPE_ARRAY>();
+ ASSERT_EQ(list_fields.size(), 2);
+ EXPECT_EQ(list_fields[0].get<TYPE_STRING>(), "alpha");
+ EXPECT_TRUE(list_fields[1].is_null());
+
+ const auto map_type = make_nullable(std::make_shared<DataTypeMap>(
+ std::make_shared<DataTypeString>(),
std::make_shared<DataTypeInt32>()));
+ const std::string binary_key_hex = "00112233445566778899aabbccddeeff0011";
Review Comment:
Addressed on the current head. The synthetic binary fixture is assembled
from shorter fragments in both V1 and V2 tests so the secret scanner no longer
classifies the test value as an API key, while the exact bytes remain covered.
##########
be/src/format/table/iceberg_reader.cpp:
##########
@@ -301,7 +301,7 @@ Status
IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
}
} else {
std::string table_column_name = old_name;
- if (const auto* table_field = _find_current_schema_field(field_id);
+ if (const auto* table_field = _find_schema_field(field_id);
Review Comment:
Addressed on the current head. V1 Parquet and ORC now carry each expanded
equality column field ID alongside its name and resolve metadata/defaults by
that ID. Tests cover drop and re-add of the same name with a different field ID.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -524,30 +560,153 @@ void enableCurrentIcebergScanSemantics() {
params.setIcebergScanSemanticsVersion(ICEBERG_SCAN_SEMANTICS_VERSION);
}
+ /**
+ * Build the schema metadata carrier used by both scanners and
equality-delete readers.
+ *
+ * <p>Batch-mode delete files are planned asynchronously after scan
parameters are sent to BE,
+ * so FE cannot first wait for their equality field IDs. Carry the latest
historical definition
+ * of every dropped primitive field; the field IDs referenced by any
planned equality delete are
+ * a subset of this bounded metadata. Iceberg field IDs are never reused.
+ */
+ @VisibleForTesting
+ List<NestedField> getSchemaFieldsForScan(Schema scanSchema) {
+ List<NestedField> fields = new ArrayList<>(scanSchema.columns());
+ if (isSystemTable) {
+ return fields;
+ }
+
+ Set<Integer> carriedFieldIds = new HashSet<>(
+ TypeUtil.indexById(scanSchema.asStruct()).keySet());
+ List<Integer> schemaIds = new
ArrayList<>(icebergTable.schemas().keySet());
+ schemaIds.sort(Collections.reverseOrder());
Review Comment:
Addressed on the current head. Historical equality fields are recovered from
the selected snapshot parent chain first and the Iceberg metadata schema list
in actual list chronology as fallback; schema IDs are never numerically sorted.
Non-monotonic schema-ID rename/drop coverage was added.
--
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]
