Skip to content

Commit b58b2e2

Browse files
OIiveirraOIiveirra
committed
[fix](iceberg) Reconcile spatial union writes and preserve binary reads
### Background Iceberg spatial writes retag proven NULLs before sink binding. Parquet files can also carry annotations that a table imports under its BINARY type. ### What problem does this PR solve? Issue Number: None Related PR: #69032 (follow-up) PR Type: Bugfix Problem Summary: Spatial NULL retagging can leave UNION ALL branches with different ordinary STRUCT field types. The spatial regression suite can expect an outer function error even when the inner MAP constructor fails first. Iceberg can validate an annotated BINARY value as spatial WKB before it reaches the binary table column. ### Details - **Frequency:** Triggered by the covered UNION writes, spatial MAP tests or annotated Iceberg BINARY imports. - **Root cause:** UNION ALL used its first branch's retagged schema, the tests assumed outer-function validation happened first, and file materialization relied on the spatial annotation. - **Why existing tests missed it:** The earlier coverage exercised like-typed UNION branches and separate typed-spatial or Hive byte paths. - **Fix:** Coerce normalized UNION branches to their common type. Assert the constructor error for spatial MAP keys. Use the raw byte type for matched Iceberg BINARY leaves in the split-local Parquet schema. - **Other paths and branches:** Keep normal spatial decoding for fields declared GEOMETRY or GEOGRAPHY. - **Behavior change:** The covered UNION writes produce a consistent schema and annotated BINARY imports preserve their payloads and NULLs. ### Testing - 96 focused FE unit tests and 211 Iceberg/Parquet BE ASAN unit tests passed. - The UNION regression failed on the previous implementation and passed with this fix. - Compiled the Groovy suite and checked 46 MAP expectations through the FE expression analyzer. - Passed Checkstyle and changed-line Clang checks. ### Release note Iceberg preserves opaque bytes in BINARY columns with spatially annotated Parquet files. Spatial NULL normalization preserves a consistent UNION schema. ### Check List (For Author) - Behavior changed: Yes (see Details) - Does this need documentation: No - Pre-submission status: Partially verified: the external Iceberg suite was not run end to end. ### AI Disclosure - Tool: OpenAI Codex - Generated: Code, tests and commit message - Human review: Pending
1 parent 5cdac8c commit b58b2e2

9 files changed

Lines changed: 276 additions & 41 deletions

File tree

‎be/src/format_v2/parquet/parquet_column_schema.cpp‎

Lines changed: 25 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -178,13 +178,13 @@ void propagate_native_max_levels(ParquetColumnSchema* schema) {
178178
}
179179
}
180180

181-
void rebuild_logical_complex_type(ParquetColumnSchema* schema) {
181+
void rebuild_logical_complex_type(ParquetColumnSchema* schema, bool force = false) {
182182
DORIS_CHECK(schema != nullptr);
183183
schema->contains_variant = schema->kind == ParquetColumnSchemaKind::VARIANT;
184184
for (const auto& child : schema->children) {
185185
schema->contains_variant |= child->contains_variant;
186186
}
187-
if (schema->kind == ParquetColumnSchemaKind::VARIANT || !schema->contains_variant) {
187+
if (schema->kind == ParquetColumnSchemaKind::VARIANT || (!force && !schema->contains_variant)) {
188188
return;
189189
}
190190

@@ -329,6 +329,29 @@ Status build_parquet_column_schema(const NativeFieldDescriptor& schema,
329329
return Status::OK();
330330
}
331331

332+
bool apply_spatial_binary_types(format::ColumnDefinition* column, ParquetColumnSchema* schema) {
333+
DORIS_CHECK(column->local_id == schema->local_id);
334+
// A table-mapped BINARY leaf may have a spatial native annotation. Update the split-local
335+
// reader schema used by both dictionary and plain-page readers, leaving the footer cache alone.
336+
const auto physical_type = remove_nullable(schema->type)->get_primitive_type();
337+
if ((physical_type == TYPE_GEOMETRY || physical_type == TYPE_GEOGRAPHY) &&
338+
remove_nullable(column->type)->get_primitive_type() == TYPE_VARBINARY) {
339+
schema->type = schema->type->is_nullable() ? column->type : remove_nullable(column->type);
340+
schema->type_descriptor.doris_type = schema->type;
341+
return true;
342+
}
343+
bool changed = false;
344+
for (size_t i = 0; i < column->children.size(); ++i) {
345+
changed |= apply_spatial_binary_types(&column->children[i], schema->children[i].get());
346+
}
347+
// A STRUCT containing an overridden leaf needs the same field type in both schemas.
348+
if (changed) {
349+
rebuild_logical_complex_type(schema, true);
350+
column->type = make_nullable(schema->type);
351+
}
352+
return changed;
353+
}
354+
332355
Status apply_variant_schema_overrides(
333356
const NativeFieldDescriptor& native_schema,
334357
const std::vector<format::LocalColumnIndex>& variant_schema_overrides,

‎be/src/format_v2/parquet/parquet_column_schema.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,8 @@ struct ParquetColumnSchema {
8484
Status build_parquet_column_schema(const NativeFieldDescriptor& schema,
8585
std::vector<std::unique_ptr<ParquetColumnSchema>>* fields);
8686

87+
bool apply_spatial_binary_types(format::ColumnDefinition* column, ParquetColumnSchema* schema);
88+
8789
Status apply_variant_schema_overrides(
8890
const NativeFieldDescriptor& native_schema,
8991
const std::vector<format::LocalColumnIndex>& variant_schema_overrides,

‎be/src/format_v2/parquet/parquet_reader.cpp‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -718,6 +718,11 @@ Status ParquetReader::get_schema(std::vector<format::ColumnDefinition>* file_sch
718718
return Status::OK();
719719
}
720720

721+
void ParquetReader::apply_spatial_binary_types(format::ColumnDefinition* file_column) {
722+
auto* schema = _state->file_schema.at(file_column->local_id).get();
723+
parquet::apply_spatial_binary_types(file_column, schema);
724+
}
725+
721726
std::unique_ptr<format::TableColumnMapper> ParquetReader::create_column_mapper(
722727
format::TableColumnMapperOptions options) const {
723728
return std::make_unique<format::ParquetColumnMapper>(std::move(options));

‎be/src/format_v2/parquet/parquet_reader.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@ class ParquetReader : public format::FileReader {
6060
void set_batch_size(size_t batch_size) override;
6161

6262
Status get_schema(std::vector<format::ColumnDefinition>* file_schema) const override;
63+
void apply_spatial_binary_types(format::ColumnDefinition* file_column);
6364

6465
std::unique_ptr<format::TableColumnMapper> create_column_mapper(
6566
format::TableColumnMapperOptions options) const override;

‎be/src/format_v2/table/iceberg_reader.cpp‎

Lines changed: 72 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -170,6 +170,12 @@ const char* file_format_name(FileFormat format) {
170170
return "UNKNOWN";
171171
}
172172

173+
bool has_spatial_file_leaf(const format::ColumnDefinition& column) {
174+
const auto type = remove_nullable(column.type)->get_primitive_type();
175+
return type == TYPE_GEOMETRY || type == TYPE_GEOGRAPHY ||
176+
std::ranges::any_of(column.children, has_spatial_file_leaf);
177+
}
178+
173179
const format::ColumnDefinition* find_matched_file_column(
174180
const format::ColumnDefinition& table_column,
175181
const std::vector<format::ColumnDefinition>& file_columns,
@@ -180,42 +186,35 @@ const format::ColumnDefinition* find_matched_file_column(
180186
: format::find_column_by_name(table_column, file_columns);
181187
}
182188

183-
// Copy the table field's timestamp marker onto the file column TableColumnMapper selects for it,
184-
// descending the way the mapper does: ARRAY and MAP children by position, STRUCT children by the
185-
// mapping mode, through an ID-less wrapper at any depth when the mapper allows one. Only a
186-
// DATETIMEV2 or TIMESTAMPTZ leaf takes the marker. A TIMESTAMP_NS leaf (an INT64 NANOS column
187-
// without the UTC annotation) keeps its nanoseconds: the Parquet reader rewrites any marked
188-
// timestamp leaf, so a marker there would make it decode TIMESTAMPTZ while the file schema still
189-
// says TIMESTAMP_NS. Returns whether any leaf took a marker.
190-
bool annotate_matched_timestamp_semantics(const format::ColumnDefinition& table_column,
191-
format::ColumnDefinition* file_column,
192-
format::TableColumnMappingMode mode,
193-
bool allow_idless_complex_wrapper_projection) {
189+
// Descend the way TableColumnMapper does: ARRAY/MAP by position and STRUCT by identity,
190+
// including ID-less wrappers when the Iceberg scan contract allows them.
191+
template <typename AnnotateLeaf>
192+
bool annotate_matched_file_leaves(const format::ColumnDefinition& table_column,
193+
format::ColumnDefinition* file_column,
194+
format::TableColumnMappingMode mode,
195+
bool allow_idless_complex_wrapper_projection,
196+
const AnnotateLeaf& annotate_leaf) {
194197
DORIS_CHECK(file_column != nullptr);
195198
DORIS_CHECK(file_column->type != nullptr);
196199
switch (remove_nullable(file_column->type)->get_primitive_type()) {
197-
case TYPE_DATETIMEV2:
198-
case TYPE_TIMESTAMPTZ:
199-
file_column->timestamp_is_adjusted_to_utc = table_column.timestamp_is_adjusted_to_utc;
200-
return file_column->timestamp_is_adjusted_to_utc.has_value();
201200
case TYPE_ARRAY:
202201
DORIS_CHECK(file_column->children.size() == 1);
203202
// FE omits nested metadata for a field it describes only by its root.
204203
return table_column.children.size() == 1 &&
205-
annotate_matched_timestamp_semantics(table_column.children[0],
206-
&file_column->children[0], mode,
207-
allow_idless_complex_wrapper_projection);
204+
annotate_matched_file_leaves(table_column.children[0], file_column->children.data(),
205+
mode, allow_idless_complex_wrapper_projection,
206+
annotate_leaf);
208207
case TYPE_MAP: {
209208
DORIS_CHECK(file_column->children.size() == 2);
210209
if (table_column.children.size() != 2) {
211210
return false;
212211
}
213-
const bool key = annotate_matched_timestamp_semantics(
214-
table_column.children[0], &file_column->children[0], mode,
215-
allow_idless_complex_wrapper_projection);
216-
const bool value = annotate_matched_timestamp_semantics(
212+
const bool key = annotate_matched_file_leaves(
213+
table_column.children[0], file_column->children.data(), mode,
214+
allow_idless_complex_wrapper_projection, annotate_leaf);
215+
const bool value = annotate_matched_file_leaves(
217216
table_column.children[1], &file_column->children[1], mode,
218-
allow_idless_complex_wrapper_projection);
217+
allow_idless_complex_wrapper_projection, annotate_leaf);
219218
return key || value;
220219
}
221220
case TYPE_STRUCT: {
@@ -225,15 +224,15 @@ bool annotate_matched_timestamp_semantics(const format::ColumnDefinition& table_
225224
const auto* matched = find_matched_file_column(table_child, file_column->children, mode,
226225
allow_idless_complex_wrapper_projection);
227226
if (matched != nullptr) {
228-
annotated |= annotate_matched_timestamp_semantics(
227+
annotated |= annotate_matched_file_leaves(
229228
table_child, &file_column->children[matched - file_column->children.data()],
230-
mode, allow_idless_complex_wrapper_projection);
229+
mode, allow_idless_complex_wrapper_projection, annotate_leaf);
231230
}
232231
}
233232
return annotated;
234233
}
235234
default:
236-
return false;
235+
return annotate_leaf(table_column, file_column);
237236
}
238237
}
239238

@@ -290,7 +289,40 @@ Status IcebergTableReader::annotate_file_schema(
290289
return Status::OK();
291290
}
292291
_apply_positional_id_fallback(file_schema);
293-
return _apply_table_timestamp_semantics(file_schema, _mapping_mode(*file_schema));
292+
const auto mode = _mapping_mode(*file_schema);
293+
// Ordinary files have no annotation to override. Avoid another field-matching pass for them.
294+
if (std::ranges::any_of(*file_schema, has_spatial_file_leaf)) {
295+
const bool allow_idless_wrappers = supports_iceberg_scan_semantics_v1(_scan_params);
296+
for (const auto& table_column : _projected_columns) {
297+
const auto* matched = find_matched_file_column(table_column, *file_schema, mode,
298+
allow_idless_wrappers);
299+
if (matched == nullptr) {
300+
continue;
301+
}
302+
auto* file_column = &(*file_schema)[matched - file_schema->data()];
303+
const auto preserve_binary = [](const format::ColumnDefinition& target,
304+
format::ColumnDefinition* file) {
305+
// An imported BINARY can retain a Parquet GEOMETRY/GEOGRAPHY annotation. Decode its
306+
// bytes without WKB validation, e.g. for a TRIANGLE unsupported by typed columns.
307+
const auto file_type = remove_nullable(file->type)->get_primitive_type();
308+
if (target.type == nullptr ||
309+
remove_nullable(target.type)->get_primitive_type() != TYPE_VARBINARY ||
310+
(file_type != TYPE_GEOMETRY && file_type != TYPE_GEOGRAPHY)) {
311+
return false;
312+
}
313+
const auto binary = remove_nullable(target.type);
314+
file->type = file->type->is_nullable() ? make_nullable(binary) : binary;
315+
return true;
316+
};
317+
if (annotate_matched_file_leaves(table_column, file_column, mode, allow_idless_wrappers,
318+
preserve_binary)) {
319+
// Apply the same types to the Parquet reader's split-local schema before mapping.
320+
static_cast<format::parquet::ParquetReader&>(*_data_reader.reader)
321+
.apply_spatial_binary_types(file_column);
322+
}
323+
}
324+
}
325+
return _apply_table_timestamp_semantics(file_schema, mode);
294326
}
295327

296328
void IcebergTableReader::_apply_positional_id_fallback(
@@ -360,8 +392,19 @@ Status IcebergTableReader::_apply_table_timestamp_semantics(
360392
auto* file_column = &(*file_schema)[matched - file_schema->data()];
361393
// Rebuilding a container recreates its type from the children, so leave every column
362394
// without a marked leaf exactly as the file reader reported it.
363-
if (annotate_matched_timestamp_semantics(table_column, file_column, mode,
364-
allow_idless_complex_wrapper_projection)) {
395+
const auto annotate_timestamp = [](const format::ColumnDefinition& table,
396+
format::ColumnDefinition* file) {
397+
const auto type = remove_nullable(file->type)->get_primitive_type();
398+
// TIMESTAMP_NS still carries nanoseconds, so it must keep its own decoder.
399+
if (type != TYPE_DATETIMEV2 && type != TYPE_TIMESTAMPTZ) {
400+
return false;
401+
}
402+
file->timestamp_is_adjusted_to_utc = table.timestamp_is_adjusted_to_utc;
403+
return file->timestamp_is_adjusted_to_utc.has_value();
404+
};
405+
if (annotate_matched_file_leaves(table_column, file_column, mode,
406+
allow_idless_complex_wrapper_projection,
407+
annotate_timestamp)) {
365408
RETURN_IF_ERROR(format::apply_file_timestamp_semantics(file_column));
366409
}
367410
}

‎be/test/format_v2/table/iceberg_reader_test.cpp‎

Lines changed: 95 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -490,7 +490,8 @@ TColumnType spatial_schema_carrier(const DataTypeSpatial& type) {
490490
void write_iceberg_spatial_key_parquet_file(const std::string& file_path,
491491
const DataTypeSpatial& spatial,
492492
const std::vector<std::optional<std::string>>& values,
493-
bool include_id, bool spatial_annotation = true) {
493+
bool include_id, bool spatial_annotation = true,
494+
bool nested = false) {
494495
::parquet::schema::NodeVector fields;
495496
if (include_id) {
496497
fields.push_back(::parquet::schema::PrimitiveNode::Make(
@@ -506,9 +507,13 @@ void write_iceberg_spatial_key_parquet_file(const std::string& file_path,
506507
: (spatial.get_primitive_type() == TYPE_GEOMETRY
507508
? ::parquet::LogicalType::Geometry(spatial.crs())
508509
: ::parquet::LogicalType::Geography(spatial.crs(), algorithm));
509-
fields.push_back(
510+
auto shape =
510511
::parquet::schema::PrimitiveNode::Make("shape", ::parquet::Repetition::OPTIONAL,
511-
annotation, ::parquet::Type::BYTE_ARRAY, -1, 1));
512+
annotation, ::parquet::Type::BYTE_ARRAY, -1, 1);
513+
fields.push_back(
514+
nested ? ::parquet::schema::GroupNode::Make("payload", ::parquet::Repetition::REQUIRED,
515+
{shape}, ::parquet::ConvertedType::NONE, 2)
516+
: shape);
512517
auto schema = std::static_pointer_cast<::parquet::schema::GroupNode>(
513518
::parquet::schema::GroupNode::Make("schema", ::parquet::Repetition::REQUIRED, fields));
514519
auto out = arrow::io::FileOutputStream::Open(file_path).ValueOrDie();
@@ -6901,6 +6906,93 @@ TEST(IcebergV2ReaderTest, UuidEqualityDeleteMatchesMissingColumnInitialDefault)
69016906
std::filesystem::remove_all(test_dir);
69026907
}
69036908

6909+
void expect_opaque_binary_rows(const Block& block, bool nested,
6910+
const std::vector<std::optional<std::string>>& values, size_t* row) {
6911+
const IColumn* shape = block.get_by_position(1).column.get();
6912+
if (nested) {
6913+
const auto& payload = assert_cast<const ColumnNullable&>(*shape);
6914+
const auto& fields = assert_cast<const ColumnStruct&>(payload.get_nested_column());
6915+
shape = &fields.get_column(0);
6916+
}
6917+
const auto& nullable = assert_cast<const ColumnNullable&>(*shape);
6918+
const auto* binary = check_and_get_column<ColumnVarbinary>(nullable.get_nested_column());
6919+
ASSERT_NE(binary, nullptr);
6920+
for (size_t i = 0; i < block.rows(); ++i, ++*row) {
6921+
ASSERT_LT(*row, values.size());
6922+
const auto& value = values[*row];
6923+
EXPECT_EQ(nullable.is_null_at(i), !value.has_value());
6924+
if (value.has_value()) {
6925+
const auto bytes = binary->get_data_at(i);
6926+
EXPECT_EQ(std::string_view(bytes.data, bytes.size), *value);
6927+
}
6928+
}
6929+
}
6930+
6931+
void check_iceberg_opaque_binary_read(const std::string& file_path,
6932+
const DataTypeSpatial& annotation, bool nested,
6933+
const std::vector<std::optional<std::string>>& values) {
6934+
ASSERT_NO_FATAL_FAILURE(write_iceberg_spatial_key_parquet_file(file_path, annotation, values,
6935+
true, true, nested));
6936+
auto scan_params = make_local_parquet_scan_params();
6937+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
6938+
auto binary = make_table_column(1, "renamed_shape", std::make_shared<DataTypeVarbinary>());
6939+
std::vector<ColumnDefinition> columns {
6940+
make_table_column(0, "id", std::make_shared<DataTypeInt32>())};
6941+
if (nested) {
6942+
auto payload = make_table_column(
6943+
2, "payload",
6944+
std::make_shared<DataTypeStruct>(DataTypes {binary.type}, Strings {binary.name}));
6945+
payload.children.push_back(std::move(binary));
6946+
columns.push_back(std::move(payload));
6947+
} else {
6948+
columns.push_back(std::move(binary));
6949+
}
6950+
6951+
RuntimeProfile profile("iceberg_binary_spatial_annotation");
6952+
RuntimeState state {TQueryOptions(), TQueryGlobals()};
6953+
io::FileReaderStats stats;
6954+
io::FileCacheStatistics cache_stats;
6955+
auto io_ctx = make_io_context(&stats, &cache_stats);
6956+
iceberg::IcebergTableReader reader;
6957+
init_iceberg_reader(&reader, columns, &scan_params, io_ctx, &state, &profile);
6958+
ASSERT_TRUE(reader.prepare_split(build_split_options(file_path)).ok());
6959+
bool eos = false;
6960+
size_t row = 0;
6961+
while (!eos) {
6962+
Block block = build_table_block(columns);
6963+
const auto status = reader.get_block(&block, &eos);
6964+
ASSERT_TRUE(status.ok()) << status;
6965+
ASSERT_NO_FATAL_FAILURE(expect_opaque_binary_rows(block, nested, values, &row));
6966+
}
6967+
EXPECT_EQ(row, values.size());
6968+
ASSERT_TRUE(reader.close().ok());
6969+
}
6970+
6971+
TEST(IcebergV2ReaderTest, BinaryColumnReadsOpaqueAnnotatedParquetWkb) {
6972+
const auto test_dir =
6973+
std::filesystem::temp_directory_path() / "doris_iceberg_binary_spatial_annotation";
6974+
std::filesystem::create_directories(test_dir);
6975+
Defer cleanup([&] { std::filesystem::remove_all(test_dir); });
6976+
// TRIANGLE is valid WKB but is outside Doris's supported typed spatial subset.
6977+
const std::string triangle(
6978+
"\x01\x11\x00\x00\x00\x01\x00\x00\x00\x04\x00\x00\x00"
6979+
"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
6980+
"\x00\x00\x00\x00\x00\x00\xf0\x3f\x00\x00\x00\x00\x00\x00\x00\x00"
6981+
"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xf0\x3f"
6982+
"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
6983+
77);
6984+
const std::vector<std::optional<std::string>> values {triangle, std::nullopt, triangle};
6985+
for (const auto& type : schema_evolution_spatial_types()) {
6986+
for (bool nested : {false, true}) {
6987+
SCOPED_TRACE(::testing::Message() << type->get_name() << " nested=" << nested);
6988+
const auto file_path =
6989+
(test_dir / (nested ? "nested.parquet" : "scalar.parquet")).string();
6990+
ASSERT_NO_FATAL_FAILURE(check_iceberg_opaque_binary_read(
6991+
file_path, assert_cast<const DataTypeSpatial&>(*type), nested, values));
6992+
}
6993+
}
6994+
}
6995+
69046996
TEST(IcebergV2ReaderTest, MissingOptionalSpatialColumnKeepsTypedNullDefault) {
69056997
const auto test_dir = std::filesystem::temp_directory_path() /
69066998
"doris_iceberg_missing_spatial_null_default_test";

0 commit comments

Comments
 (0)