github-actions[bot] commented on code in PR #66538:
URL: https://github.com/apache/doris/pull/66538#discussion_r3851594513
##########
be/src/format/table/iceberg_reader.cpp:
##########
@@ -183,6 +486,9 @@ Status IcebergTableReader::get_next_block_inner(Block*
block, size_t* read_rows,
RETURN_IF_ERROR(_expand_block_if_need(block));
RETURN_IF_ERROR(_file_format_reader->get_next_block(block, read_rows,
eof));
+ RETURN_IF_ERROR(_materialize_missing_table_columns(block, *read_rows));
Review Comment:
[P1] Materialize whole-column defaults before V1 filtering
For a struct/list/map that is entirely absent from an old file, FE gives the
physical reader a NULL placeholder and relies on this call to install the real
Iceberg initial default. However, both V1 Parquet and ORC evaluate
missing-column conjuncts inside `get_next_block()` before control reaches this
line. Predicates such as `added_struct IS NOT NULL` or a nested comparison
therefore discard valid rows based on NULL and cannot be repaired afterward.
The real typed default must reach the physical reader, or these conjuncts must
be deferred until after materialization.
##########
be/src/format/table/iceberg_reader.cpp:
##########
@@ -487,134 +1189,140 @@ Status IcebergParquetReader::init_reader(
parquet_reader->set_row_lineage_columns(_row_lineage_columns);
}
- auto column_id_result = _create_column_ids(_data_file_field_desc,
tuple_descriptor);
- auto& column_ids = column_id_result.column_ids;
- const auto& filter_column_ids = column_id_result.filter_column_ids;
-
- RETURN_IF_ERROR(init_row_filters());
_all_required_col_names = file_col_names;
+ for (const auto* slot : tuple_descriptor->slots()) {
+ _id_to_block_column_name.emplace(slot->col_unique_id(),
slot->col_name());
+ }
+ RETURN_IF_ERROR(init_row_filters());
if (!_params.__isset.history_schema_info ||
_params.history_schema_info.empty()) [[unlikely]] {
RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_name(
tuple_descriptor, *_data_file_field_desc,
table_info_node_ptr));
} else {
- std::set<std::string> read_col_name_set(file_col_names.begin(),
file_col_names.end());
+
RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_field_id_with_name_mapping(
+ _params.history_schema_info.front().root_field,
*_data_file_field_desc,
+ table_info_node_ptr,
supports_iceberg_scan_semantics_v2(&_params)));
+ }
- bool exist_field_id = true;
- for (int idx = 0; idx < _data_file_field_desc->size(); idx++) {
- if (_data_file_field_desc->get_column(idx)->field_id == -1) {
- // the data file may be from hive table migrated to iceberg,
field id is missing
- exist_field_id = false;
- break;
- }
- }
- const auto& table_schema =
_params.history_schema_info.front().root_field;
-
- table_info_node_ptr =
std::make_shared<TableSchemaChangeHelper::StructNode>();
- if (exist_field_id) {
- // id -> table column name. columns that need read data file.
- std::unordered_map<int, std::shared_ptr<schema::external::TField>>
id_to_table_field;
- for (const auto& table_field : table_schema.fields) {
- auto field = table_field.field_ptr;
- DCHECK(field->__isset.name);
- if (!read_col_name_set.contains(field->name)) {
- continue;
- }
- id_to_table_field.emplace(field->id, field);
- }
+ auto column_id_result =
+ _create_column_ids(_data_file_field_desc, tuple_descriptor,
table_info_node_ptr);
+ auto& column_ids = column_id_result.column_ids;
+ const auto& filter_column_ids = column_id_result.filter_column_ids;
- for (int idx = 0; idx < _data_file_field_desc->size(); idx++) {
- const auto& data_file_field =
_data_file_field_desc->get_column(idx);
- auto data_file_column_id =
_data_file_field_desc->get_column(idx)->field_id;
-
- if (id_to_table_field.contains(data_file_column_id)) {
- const auto& table_field =
id_to_table_field[data_file_column_id];
-
- std::shared_ptr<TableSchemaChangeHelper::Node> field_node
= nullptr;
- RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_field_id(
- *table_field, *data_file_field, exist_field_id,
field_node));
- table_info_node_ptr->add_children(table_field->name,
data_file_field->name,
- field_node);
-
- _id_to_block_column_name.emplace(data_file_column_id,
table_field->name);
- id_to_table_field.erase(data_file_column_id);
- } else if
(_equality_delete_col_ids.contains(data_file_column_id)) {
- // Columns that need to be read for equality delete.
- const static std::string EQ_DELETE_PRE =
"__equality_delete_column__";
-
- // Construct table column names that avoid duplication
with current table schema.
- // As the columns currently being read may have been
deleted in the latest
- // table structure or have undergone a series of schema
changes...
- std::string table_column_name = EQ_DELETE_PRE +
data_file_field->name;
- table_info_node_ptr->add_children(
- table_column_name, data_file_field->name,
-
std::make_shared<TableSchemaChangeHelper::ConstNode>());
-
- _id_to_block_column_name.emplace(data_file_column_id,
table_column_name);
- _expand_col_names.emplace_back(table_column_name);
- auto expand_data_type =
make_nullable(data_file_field->data_type);
- _expand_columns.emplace_back(
- ColumnWithTypeAndName
{expand_data_type->create_column(),
- expand_data_type,
table_column_name});
-
- _all_required_col_names.emplace_back(table_column_name);
- column_ids.insert(data_file_field->get_column_id());
+ const static std::string EQ_DELETE_PRE = "__equality_delete_column__";
+ bool all_file_columns_have_field_ids = true;
+ bool any_file_column_has_field_id = false;
+ for (int index = 0; index < _data_file_field_desc->size(); ++index) {
+ const auto* field = _data_file_field_desc->get_column(index);
+ if (field == nullptr) {
+ continue;
+ }
+ if (field->field_id < 0) {
+ all_file_columns_have_field_ids = false;
+ }
+ if (parquet_subtree_has_iceberg_id(*field)) {
+ any_file_column_has_field_id = true;
+ }
+ }
+ const bool use_field_ids = supports_iceberg_scan_semantics_v2(&_params)
+ ? any_file_column_has_field_id
+ : all_file_columns_have_field_ids;
+ std::vector<std::string> new_expand_col_names;
+ DORIS_CHECK(_expand_col_names.size() == _expand_col_field_ids.size());
+ DORIS_CHECK(_expand_col_names.size() == _expand_columns.size());
+ for (size_t index = 0; index < _expand_col_names.size(); ++index) {
+ const std::string old_name = _expand_col_names[index];
+ const int32_t field_id = _expand_col_field_ids[index];
+ const FieldSchema* file_column = nullptr;
+ ParquetEqualityFieldPath file_path;
+ bool complete_file_path = false;
+ if (use_field_ids) {
+ complete_file_path =
find_parquet_equality_field_path_by_id(_data_file_field_desc,
+
field_id, &file_path);
+ if (!complete_file_path &&
supports_iceberg_scan_semantics_v2(&_params)) {
+ const auto table_path = _find_schema_field_path(field_id);
+ if (!table_path.empty()) {
+ complete_file_path =
find_parquet_equality_field_prefix_by_id_path(
+ _data_file_field_desc, table_path, &file_path);
}
}
- for (const auto& [id, table_field] : id_to_table_field) {
- table_info_node_ptr->add_not_exist_children(table_field->name);
+ if (!file_path.fields.empty()) {
+ file_column = file_path.fields.front();
}
} else {
- if (!_equality_delete_col_ids.empty()) [[unlikely]] {
- return Status::InternalError(
- "Can not read missing field id data file when have
equality delete");
- }
- std::map<std::string, size_t> file_column_idx_map;
- for (size_t idx = 0; idx < _data_file_field_desc->size(); idx++) {
-
file_column_idx_map.emplace(_data_file_field_desc->get_column(idx)->name, idx);
- }
-
- for (const auto& table_field : table_schema.fields) {
- DCHECK(table_field.__isset.field_ptr);
- DCHECK(table_field.field_ptr->__isset.name);
- const auto& table_column_name = table_field.field_ptr->name;
- if (!read_col_name_set.contains(table_column_name)) {
- continue;
- }
- if (!table_field.field_ptr->__isset.name_mapping ||
- table_field.field_ptr->name_mapping.size() == 0) {
- return Status::DataQualityError(
- "name_mapping must be set when read missing field
id data file.");
- }
- bool have_mapping = false;
- for (const auto& mapped_name :
table_field.field_ptr->name_mapping) {
- if (file_column_idx_map.contains(mapped_name)) {
- std::shared_ptr<TableSchemaChangeHelper::Node>
field_node = nullptr;
- const auto& file_field =
_data_file_field_desc->get_column(
- file_column_idx_map.at(mapped_name));
-
RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_field_id(
- *table_field.field_ptr, *file_field,
exist_field_id, field_node));
- table_info_node_ptr->add_children(table_column_name,
file_field->name,
- field_node);
- have_mapping = true;
- break;
- }
- }
- if (!have_mapping) {
-
table_info_node_ptr->add_not_exist_children(table_column_name);
+ const auto table_path = _find_schema_field_path(field_id);
+ if (!table_path.empty()) {
+ complete_file_path =
find_parquet_equality_field_prefix_by_name_path(
+ _data_file_field_desc, table_path, old_name,
&file_path);
+ if (!file_path.fields.empty()) {
+ file_column = file_path.fields.front();
}
}
}
+
+ const std::string leaf_name =
+ file_path.fields.empty() ? old_name :
file_path.fields.back()->name;
+ const std::string block_name = EQ_DELETE_PRE +
std::to_string(field_id) + "_" + leaf_name;
+ _id_to_block_column_name[field_id] = block_name;
+ _expand_columns[index].name = block_name;
+ new_expand_col_names.push_back(block_name);
+ if (file_column == nullptr) {
+ RETURN_IF_ERROR(_register_missing_equality_delete_column(field_id,
block_name,
+
_expand_columns[index].type));
+ continue;
+ }
+ if (!complete_file_path) {
+ ColumnPtr missing_value;
+ RETURN_IF_ERROR(
+ _create_missing_equality_delete_value(field_id,
_expand_columns[index].type,
+
file_path.fields.size(), &missing_value));
+ _nested_equality_delete_columns.push_back({
+ .field_id = field_id,
+ .block_name = block_name,
+ .source_leaf_type = _expand_columns[index].type,
+ .leaf_type = _expand_columns[index].type,
+ .child_indexes = file_path.child_indexes,
+ .missing_value = std::move(missing_value),
+ .cast_context = nullptr,
+ });
+ RETURN_IF_ERROR(_prepare_nested_equality_delete_column(
+ &_nested_equality_delete_columns.back()));
+ _expand_columns[index].type =
make_nullable(file_column->data_type);
+ _expand_columns[index].column =
_expand_columns[index].type->create_column();
+ } else if (!file_path.child_indexes.empty()) {
+ _nested_equality_delete_columns.push_back({
+ .field_id = field_id,
+ .block_name = block_name,
+ .source_leaf_type =
make_nullable(file_path.fields.back()->data_type),
+ .leaf_type = _expand_columns[index].type,
+ .child_indexes = file_path.child_indexes,
+ .missing_value = nullptr,
+ .cast_context = nullptr,
+ });
+ RETURN_IF_ERROR(_prepare_nested_equality_delete_column(
+ &_nested_equality_delete_columns.back()));
+ _expand_columns[index].type =
make_nullable(file_column->data_type);
+ _expand_columns[index].column =
_expand_columns[index].type->create_column();
+ }
+ for (uint64_t column_id = file_column->get_column_id();
+ column_id <= file_column->get_max_column_id(); ++column_id) {
+ column_ids.insert(column_id);
+ }
+ _all_required_col_names.push_back(block_name);
+ table_info_node_ptr->add_children(block_name, file_column->name,
Review Comment:
[P1] Share the physical root across nested equality keys
This adds a new table alias for every nested equality key, but
`ParquetReader::init_reader()` reverses these mappings with
`required_file_columns.emplace(fileName, tableName)`, so one physical root can
populate only the first alias. If the struct is projected normally, that slot
wins; if two delete keys share the struct, only one hidden alias wins. The
other hidden column remains empty and equality deletes are missed (or fail on
the size mismatch). Please read each physical root once and extract all nested
keys from that populated column.
##########
be/src/format/transformer/vorc_transformer.cpp:
##########
@@ -596,6 +672,195 @@ Status VOrcTransformer::write(const Block& block) {
return Status::OK();
}
+static Status normalize_iceberg_nullable_column(const ColumnPtr& column, const
DataTypePtr& type,
+ const iceberg::NestedField&
nested_field,
+ ColumnPtr* normalized_column,
+ const NullMap* skipped_rows) {
+ const auto& nullable_column = assert_cast<const ColumnNullable&>(*column);
+ const auto& null_map = nullable_column.get_null_map_data();
+ NullMap combined_null_map;
+ const NullMap* combined_skipped_rows = &null_map;
+ if (skipped_rows != nullptr) {
+ combined_null_map.resize(null_map.size());
+ for (size_t row = 0; row < null_map.size(); ++row) {
+ combined_null_map[row] = null_map[row] | (*skipped_rows)[row];
+ }
+ combined_skipped_rows = &combined_null_map;
+ }
+ ColumnPtr nested_column;
+
RETURN_IF_ERROR(normalize_iceberg_binary_column(nullable_column.get_nested_column_ptr(),
+ remove_nullable(type),
nested_field,
+ &nested_column,
combined_skipped_rows));
+ *normalized_column = ColumnNullable::create(
+ IColumn::mutate(std::move(nested_column)),
+
IColumn::mutate(nullable_column.get_null_map_column_ptr()->clone()));
+ return Status::OK();
+}
+
+static Status normalize_iceberg_uuid_column(const ColumnPtr& column,
ColumnPtr* normalized_column,
+ const NullMap* skipped_rows) {
+ DORIS_CHECK(check_and_get_column<ColumnString>(*column) != nullptr ||
+ check_and_get_column<ColumnVarbinary>(*column) != nullptr);
+ auto binary_column = column->clone_empty();
+ binary_column->reserve(column->size());
+ for (size_t row = 0; row < column->size(); ++row) {
+ std::array<uint8_t, 16> bytes;
+ if (skipped_rows == nullptr || (*skipped_rows)[row] == 0) {
+
RETURN_IF_ERROR(parse_iceberg_uuid_to_bytes(column->get_data_at(row), &bytes));
+ } else {
+ bytes.fill(0);
+ }
+ binary_column->insert_data(reinterpret_cast<const
char*>(bytes.data()), bytes.size());
+ }
+ *normalized_column = std::move(binary_column);
+ return Status::OK();
+}
+
+static Status normalize_iceberg_fixed_column(const ColumnPtr& column,
+ const iceberg::NestedField&
nested_field,
+ ColumnPtr* normalized_column,
+ const NullMap* skipped_rows) {
+ const auto expected_length = cast_set<size_t>(
+ assert_cast<const
iceberg::FixedType*>(nested_field.field_type())->get_length());
+ for (size_t row = 0; row < column->size(); ++row) {
+ if (skipped_rows != nullptr && (*skipped_rows)[row] != 0) {
+ continue;
+ }
+ const auto value = column->get_data_at(row);
+ if (value.size != expected_length) {
Review Comment:
[P1] Pad legacy CHAR values for Iceberg FIXED
With varbinary mapping disabled, Iceberg FIXED[n] is exposed as Doris
CHAR(n). A short CHAR payload is normal and the Parquet fixed-size writer
zero-pads it to n bytes, but this ORC path rejects every value whose stored
`StringRef` is shorter than n because it does not receive the Doris type. Valid
inserts therefore fail only when the table writes ORC. Preserve the existing
TYPE_CHAR padding contract while keeping exact-length validation for
binary/string carriers.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/IcebergUpdateCommand.java:
##########
@@ -324,12 +344,17 @@ public Plan getExplainPlan(ConnectContext ctx) {
}
IcebergExternalTable icebergTable = (IcebergExternalTable) table;
IcebergDmlCommandUtils.checkUpdateMode(icebergTable);
+ IcebergWriteSchemaContext writeSchemaContext =
IcebergWriteSchemaContext.create(
+ icebergTable, Optional.empty());
+ Optional<IcebergWriteSchemaContext> previousWriteSchemaContext =
+ IcebergDmlCommandUtils.installWriteSchemaContext(ctx,
writeSchemaContext);
long previousTargetTableId = ctx.getIcebergRowIdTargetTableId();
ctx.setIcebergRowIdTargetTableId(table.getId());
try {
- return buildMergePlan(ctx, logicalQuery, assignments,
icebergTable);
+ return buildMergePlan(ctx, logicalQuery, assignments,
icebergTable, writeSchemaContext);
} finally {
ctx.setIcebergRowIdTargetTableId(previousTargetTableId);
+ IcebergDmlCommandUtils.restoreWriteSchemaContext(ctx,
previousWriteSchemaContext);
Review Comment:
[P2] Keep the pinned context through EXPLAIN analysis
`getExplainPlan()` returns an un-analyzed merge tree, but this restores the
Iceberg write context before `ExplainCommand` calls `planner.plan()`.
Assignment/value DEFAULTs happen to be rewritten eagerly;
DEFAULT(target_column) in UPDATE WHERE, MERGE ON, or action predicates reaches
`RewriteDefaultExpression` later and fails because the pinned target is gone,
although executing the same DML succeeds. Carry the context through explain
planning or eagerly rewrite every DEFAULT occurrence before restoring it.
##########
be/src/format/table/iceberg_reader.cpp:
##########
@@ -988,134 +1729,141 @@ Status IcebergTableReader::read_deletion_vector(const
std::string& data_file_pat
// attributes/column IDs, it is not easy to combine them.
Status IcebergParquetReader::_process_equality_delete(
const std::vector<TIcebergDeleteFileDesc>& delete_files) {
+ struct ReadSpec {
+ NestedEqualityDeleteColumn nested_field;
+ std::string root_name;
+ DataTypePtr root_type;
+ };
std::unordered_map<std::string, std::tuple<std::string, const
SlotDescriptor*>>
partition_columns;
std::unordered_map<std::string, VExprContextSPtr> missing_columns;
- std::map<int, const FieldSchema*> data_file_id_to_field_schema;
- for (int idx = 0; idx < _data_file_field_desc->size(); ++idx) {
- auto field_schema = _data_file_field_desc->get_column(idx);
- if (_data_file_field_desc->get_column(idx)->field_id == -1) {
- return Status::DataQualityError("Iceberg equality delete data file
missing field id.");
- }
-
data_file_id_to_field_schema[_data_file_field_desc->get_column(idx)->field_id] =
- field_schema;
- }
-
for (const auto& delete_file : delete_files) {
+ if (!delete_file.__isset.field_ids) [[unlikely]] {
+ return Status::InternalError(
+ "missing delete field ids when reading equality delete
file");
+ }
TFileRangeDesc delete_desc;
- // must use __set() method to make sure __isset is true
delete_desc.__set_fs_name(_range.fs_name);
delete_desc.path = delete_file.path;
delete_desc.start_offset = 0;
delete_desc.size = -1;
delete_desc.file_size = -1;
- if (!delete_file.__isset.field_ids) [[unlikely]] {
- return Status::InternalError(
- "missing delete field ids when reading equality delete
file");
- }
- auto& read_column_field_ids = delete_file.field_ids;
- std::set<int> read_column_field_ids_set;
- for (const auto& field_id : read_column_field_ids) {
- read_column_field_ids_set.insert(field_id);
- _equality_delete_col_ids.insert(field_id);
- }
-
auto delete_reader = ParquetReader::create_unique(
_profile, _params, delete_desc, READ_DELETE_FILE_BATCH_SIZE,
const_cast<cctz::time_zone*>(&_state->timezone_obj()),
_io_ctx, _state,
_meta_cache);
RETURN_IF_ERROR(delete_reader->init_schema_reader());
+ const FieldDescriptor* delete_field_desc = nullptr;
+
RETURN_IF_ERROR(delete_reader->get_file_metadata_schema(&delete_field_desc));
+ DORIS_CHECK(delete_field_desc != nullptr);
- // the column that to read equality delete file.
- // (delete file may be have extra columns that don't need to read)
+ std::vector<ReadSpec> read_specs;
std::vector<std::string> delete_col_names;
std::vector<DataTypePtr> delete_col_types;
std::vector<int> delete_col_ids;
- std::unordered_map<std::string, uint32_t> delete_col_name_to_block_idx;
-
- const FieldDescriptor* delete_field_desc = nullptr;
-
RETURN_IF_ERROR(delete_reader->get_file_metadata_schema(&delete_field_desc));
- DCHECK(delete_field_desc != nullptr);
-
+ std::vector<std::string> read_root_names;
+ std::vector<DataTypePtr> read_root_types;
+ std::unordered_map<std::string, uint32_t> read_root_positions;
auto eq_file_node =
std::make_shared<TableSchemaChangeHelper::StructNode>();
- for (const auto& delete_file_field :
delete_field_desc->get_fields_schema()) {
- if (delete_file_field.field_id == -1) [[unlikely]] { // missing
delete_file_field id
- // equality delete file must have delete_file_field id to
match column.
+ for (int32_t field_id : delete_file.field_ids) {
+ ParquetEqualityFieldPath path;
+ if (!find_parquet_equality_field_path_by_id(delete_field_desc,
field_id, &path)) {
return Status::DataQualityError(
- "missing delete_file_field id when reading equality
delete file");
- } else if
(read_column_field_ids_set.contains(delete_file_field.field_id)) {
- // the column that need to read.
- if (delete_file_field.children.size() > 0) [[unlikely]] { //
complex column
- return Status::InternalError(
- "can not support read complex column in equality
delete file");
- } else if
(!data_file_id_to_field_schema.contains(delete_file_field.field_id))
- [[unlikely]] {
- return Status::DataQualityError(
- "can not find delete field id in data file schema
when reading "
- "equality delete file");
- }
- auto data_file_field =
data_file_id_to_field_schema[delete_file_field.field_id];
- if (data_file_field->data_type->get_primitive_type() !=
- delete_file_field.data_type->get_primitive_type())
[[unlikely]] {
- return Status::NotSupported(
- "Not Support type change in equality delete,
field: {}, delete "
- "file type: {}, data file type: {}",
- delete_file_field.field_id,
delete_file_field.data_type->get_name(),
- data_file_field->data_type->get_name());
- }
-
- std::string filed_lower_name =
to_lower(delete_file_field.name);
- eq_file_node->add_children(filed_lower_name,
delete_file_field.name,
-
std::make_shared<TableSchemaChangeHelper::ScalarNode>());
-
- delete_col_ids.emplace_back(delete_file_field.field_id);
- delete_col_names.emplace_back(filed_lower_name);
-
delete_col_types.emplace_back(make_nullable(delete_file_field.data_type));
-
- read_column_field_ids_set.erase(delete_file_field.field_id);
- } else {
- // delete file may be have extra columns that don't need to
read
+ "missing field id {} when reading equality delete file
{}", field_id,
+ delete_file.path);
+ }
+ DORIS_CHECK(!path.fields.empty());
+ const auto* root = path.fields.front();
+ const auto* leaf = path.fields.back();
+ if (!leaf->children.empty()) {
+ return Status::NotSupported(
+ "Iceberg equality delete does not support complex
column {}", leaf->name);
+ }
+ const std::string leaf_name = to_lower(leaf->name);
+ const std::string root_name = to_lower(root->name);
+ const auto leaf_type = make_nullable(leaf->data_type);
+ read_specs.push_back({
+ {
+ .field_id = field_id,
+ .block_name = leaf_name,
+ .source_leaf_type = leaf_type,
+ .leaf_type = leaf_type,
+ .child_indexes = path.child_indexes,
+ .missing_value = nullptr,
+ .cast_context = nullptr,
+ },
+ root_name,
+ make_nullable(root->data_type),
+ });
+ delete_col_ids.push_back(field_id);
+ delete_col_names.push_back(leaf_name);
+ delete_col_types.push_back(leaf_type);
+ _equality_delete_col_ids.insert(field_id);
+ if (!_id_to_block_column_name.contains(field_id) &&
Review Comment:
[P1] Preserve the historical type for each equality delete
This suppresses the delete-typed hidden carrier whenever the field is
already projected. After a legal INT-to-BIGINT promotion, one old INT
equality-delete file is therefore probed with the current BIGINT data column:
the multi-key path rejects the type mismatch, and the simple-key HybridSet hits
its exact `assert_cast`, failing the scan. Grouping multiple delete files only
by their ID vector has the same problem when it tries to merge old and new key
types. Keep predicates/carriers per delete schema and cast the data key to each
historical type before probing.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -592,25 +624,85 @@ private String getDeleteFileContentType(int content) {
}
public void createScanRangeLocations() throws UserException {
- super.createScanRangeLocations();
+ Schema scanSchema = getQuerySchema();
+ Optional<Map<Integer, List<String>>> nameMapping =
extractNameMapping();
+ Set<Integer> equalityDeleteFieldIds = Collections.emptySet();
+ if (!isSystemTable) {
+ ConnectContext context =
Preconditions.checkNotNull(ConnectContext.get(),
+ "Connect context is required for Iceberg scan planning");
+ checkFileScannerV1BackendCompatibility(
+ context.getSessionVariable().enableFileScannerV2,
backendPolicy.getBackends());
+ boolean batchMode = isBatchMode();
+ boolean batchMayHaveEqualityDeletes = batchMode &&
mayHaveEqualityDeletes();
Review Comment:
[P1] Preserve lazy planning on current-only backends
This calls `mayHaveEqualityDeletes()` for every batch scan. Unless the
snapshot summary proves zero, that method synchronously invokes
`scan.planFiles()` and can exhaust the full filtered task set before
`super.createScanRangeLocations()` starts the lazy producer, which plans the
scan again. On large tables this defeats batch planning even when every
selected backend supports the new semantics. Gate this exact preflight on a
smooth-upgrade source backend (or otherwise avoid planning tasks twice).
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -620,30 +712,904 @@ void enableCurrentIcebergScanSemantics() {
params.setIcebergScanSemanticsVersion(ICEBERG_SCAN_SEMANTICS_VERSION);
}
+ /**
+ * Build the schema metadata carrier used by both scanners and
equality-delete readers.
+ *
+ * <p>Non-batch scans preplan their exact tasks, so their equality field
IDs can extend the
+ * query-wide carrier here. Batch scans keep planning lazy and attach any
required historical
+ * fragments to the exact split instead. In both modes, unrelated dropped
types stay outside the
+ * serialized schema.
+ */
+ @VisibleForTesting
+ List<NestedField> getSchemaFieldsForScan(
+ Schema scanSchema, Set<Integer> equalityDeleteFieldIds) throws
UserException {
+ List<NestedField> fields = new ArrayList<>(scanSchema.columns());
+ if (isSystemTable || equalityDeleteFieldIds.isEmpty()) {
+ return fields;
+ }
+
+ Set<Integer> missingFieldIds = new HashSet<>(equalityDeleteFieldIds);
+
missingFieldIds.removeAll(TypeUtil.indexById(scanSchema.asStruct()).keySet());
+ if (missingFieldIds.isEmpty()) {
+ return fields;
+ }
+
+ List<Schema> schemaHistory = getMetadataSchemaHistory();
+ // Schema IDs may be reused when evolution returns to an earlier
schema, while the metadata
+ // list may also contain schemas committed after a time-travel or
branch target. Follow the
+ // actual scan snapshot's parent chain first so the field definition
active on that lineage
+ // wins. Then use the complete metadata list as a fallback for
schema-only changes and
+ // expired ancestors. A fallback definition may come from a later
rename, so BE resolves an
+ // ID-less equality key through the target mapping first and the
delete file's original key
+ // name second. Initial-default and field identity remain bound to the
stable field ID.
+ Snapshot snapshot = createTableScan().snapshot();
+ while (snapshot != null) {
+ Integer schemaId = snapshot.schemaId();
+ if (schemaId != null) {
+ Schema historicalSchema = icebergTable.schemas().get(schemaId);
+ Preconditions.checkState(historicalSchema != null,
+ "Iceberg snapshot schema %s is absent from table
metadata", schemaId);
+ addHistoricalEqualityFields(fields, missingFieldIds,
historicalSchema);
+ }
+ Long parentId = snapshot.parentId();
+ snapshot = parentId == null ? null :
icebergTable.snapshot(parentId);
+ }
+ for (int index = schemaHistory.size() - 1; index >= 0; index--) {
+ addHistoricalEqualityFields(fields, missingFieldIds,
schemaHistory.get(index));
+ }
+ Preconditions.checkState(missingFieldIds.isEmpty(),
+ "Iceberg equality-delete fields are absent from schema
history: %s",
+ missingFieldIds);
+ return fields;
+ }
+
+ private TSchema getEqualityDeleteSchema(Set<Integer>
equalityDeleteFieldIds)
+ throws UserException {
+ if (equalityDeleteFieldIds.isEmpty()) {
+ return null;
+ }
+ Preconditions.checkState(plannedScanSchema != null,
+ "Iceberg scan schema must be pinned before split planning");
+ Set<Integer> missingFieldIds = new HashSet<>(equalityDeleteFieldIds);
+
missingFieldIds.removeAll(TypeUtil.indexById(plannedScanSchema.asStruct()).keySet());
+ if (missingFieldIds.isEmpty()) {
+ return null;
+ }
+ Set<Integer> cacheKey = Collections.unmodifiableSet(new
HashSet<>(missingFieldIds));
+ TSchema cached = equalityDeleteSchemaCache.get(cacheKey);
+ if (cached != null) {
+ return cached;
+ }
+
+ Schema mergedSchema = new Schema(
+ getSchemaFieldsForScan(plannedScanSchema, missingFieldIds));
+ List<NestedField> selectedFields = TypeUtil.select(mergedSchema,
missingFieldIds).columns();
+ List<Column> selectedColumns = new ArrayList<>();
+ for (NestedField field : selectedFields) {
+ selectedColumns.add(IcebergUtils.parseField(
+ field, getEnableMappingVarbinary(),
getEnableMappingTimestampTz()));
+ }
+ TSchema schema = ExternalUtil.createSchemaInfoForAllColumn(
+ -1L, selectedColumns,
plannedNameMapping.orElse(Collections.emptyMap()),
+ plannedNameMapping.isPresent(),
+ IcebergUtils.getSerializedInitialDefaults(
+ selectedFields, getEnableMappingTimestampTz()),
+ IcebergUtils.getBinaryLikeFieldIds(selectedFields),
+ IcebergUtils.getRequiredFieldIds(selectedFields));
+ equalityDeleteSchemaCache.put(cacheKey, schema);
+ return schema;
+ }
+
+ private List<Schema> getMetadataSchemaHistory() {
+ Preconditions.checkState(icebergTable instanceof HasTableOperations,
+ "Iceberg table does not expose metadata schema history: %s",
icebergTable.name());
+ return ((HasTableOperations)
icebergTable).operations().current().schemas();
+ }
+
+ /**
+ * Return only schemas that can describe files visible from the selected
target.
+ *
+ * <p>The query schema is included explicitly because a schema-only commit
does not create a
+ * snapshot. Other schemas are taken from the selected snapshot's parent
lineage and from
+ * cherry-picked source snapshots (including their ancestry), excluding
later main-branch and
+ * unrelated branch schemas from the rolling-upgrade fence. An empty
optional means snapshot
+ * expiration truncated any required lineage, so callers must
conservatively require current
+ * scan semantics.
+ */
+ @VisibleForTesting
+ Optional<List<Schema>> getRequiredFieldSchemaHistory(Schema scanSchema)
throws UserException {
+ List<Schema> schemas = new ArrayList<>();
+ Set<Integer> schemaIds = new HashSet<>();
+ schemas.add(scanSchema);
+ schemaIds.add(scanSchema.schemaId());
+
+ Snapshot selectedSnapshot = createTableScan().snapshot();
+ Deque<Snapshot> snapshots = new ArrayDeque<>();
+ if (selectedSnapshot != null) {
+ snapshots.add(selectedSnapshot);
+ }
+ Set<Long> visitedSnapshotIds = new HashSet<>();
+ while (!snapshots.isEmpty()) {
+ Snapshot snapshot = snapshots.removeFirst();
+ if (!visitedSnapshotIds.add(snapshot.snapshotId())) {
+ continue;
+ }
+ Integer schemaId = snapshot.schemaId();
+ if (schemaId != null && schemaIds.add(schemaId)) {
+ Schema lineageSchema = icebergTable.schemas().get(schemaId);
+ Preconditions.checkState(lineageSchema != null,
+ "Iceberg snapshot schema %s is absent from table
metadata", schemaId);
+ schemas.add(lineageSchema);
+ }
+ Long parentId = snapshot.parentId();
+ if (parentId != null) {
+ Snapshot parent = icebergTable.snapshot(parentId);
+ if (parent == null) {
+ return Optional.empty();
+ }
+ snapshots.addLast(parent);
+ }
+ String sourceSnapshotId =
+
snapshot.summary().get(SnapshotSummary.SOURCE_SNAPSHOT_ID_PROP);
+ if (sourceSnapshotId != null) {
+ Snapshot sourceSnapshot =
+
icebergTable.snapshot(Long.parseLong(sourceSnapshotId));
+ if (sourceSnapshot == null) {
+ return Optional.empty();
+ }
+ snapshots.addLast(sourceSnapshot);
+ }
+ }
+ return Optional.of(schemas);
+ }
+
+ private static void addHistoricalEqualityFields(List<NestedField> fields,
+ Set<Integer> missingFieldIds, Schema historicalSchema) {
+ Map<Integer, NestedField> historicalFields =
+ TypeUtil.indexById(historicalSchema.asStruct());
+ Set<Integer> selectedFieldIds = new HashSet<>();
+ for (Integer fieldId : missingFieldIds) {
+ NestedField field = historicalFields.get(fieldId);
+ if (field != null) {
+ Preconditions.checkState(field.type().isPrimitiveType(),
+ "Iceberg equality-delete field %s must be primitive",
fieldId);
+ selectedFieldIds.add(fieldId);
+ }
+ }
+ if (selectedFieldIds.isEmpty()) {
+ return;
+ }
+
+ Schema selectedSchema = TypeUtil.select(historicalSchema,
selectedFieldIds);
+ mergeHistoricalEqualityFields(fields, selectedSchema.columns());
+ missingFieldIds.removeAll(selectedFieldIds);
+ }
+
+ private static void mergeHistoricalEqualityFields(
+ List<NestedField> fields, List<NestedField> historicalFields) {
+ for (NestedField historicalField : historicalFields) {
+ int currentIndex = -1;
+ for (int index = 0; index < fields.size(); index++) {
+ if (fields.get(index).fieldId() == historicalField.fieldId()) {
+ currentIndex = index;
+ break;
+ }
+ }
+ if (currentIndex < 0) {
+ fields.add(historicalField);
+ continue;
+ }
+
+ NestedField currentField = fields.get(currentIndex);
+ Type mergedType = mergeHistoricalEqualityType(
+ currentField.type(), historicalField.type());
+ if (mergedType != currentField.type()) {
+ fields.set(currentIndex, Types.NestedField.from(currentField)
+ .ofType(mergedType)
+ .build());
+ }
+ }
+ }
+
+ private static Type mergeHistoricalEqualityType(Type currentType, Type
historicalType) {
+ Preconditions.checkState(currentType.typeId() ==
historicalType.typeId(),
+ "Iceberg equality-delete ancestor type changed from %s to %s",
+ historicalType, currentType);
+ switch (currentType.typeId()) {
+ case STRUCT:
+ List<NestedField> mergedFields =
+ new ArrayList<>(currentType.asStructType().fields());
+ mergeHistoricalEqualityFields(
+ mergedFields, historicalType.asStructType().fields());
+ if (mergedFields.equals(currentType.asStructType().fields())) {
+ return currentType;
+ }
+ return Types.StructType.of(mergedFields);
+ case LIST:
+ Types.ListType currentList = currentType.asListType();
+ Types.ListType historicalList = historicalType.asListType();
+ Preconditions.checkState(currentList.elementId() ==
historicalList.elementId(),
+ "Iceberg equality-delete list element id changed from
%s to %s",
+ historicalList.elementId(), currentList.elementId());
+ Type mergedElement = mergeHistoricalEqualityType(
+ currentList.elementType(),
historicalList.elementType());
+ if (mergedElement == currentList.elementType()) {
+ return currentType;
+ }
+ return currentList.isElementOptional()
+ ? Types.ListType.ofOptional(currentList.elementId(),
mergedElement)
+ : Types.ListType.ofRequired(currentList.elementId(),
mergedElement);
+ case MAP:
+ Types.MapType currentMap = currentType.asMapType();
+ Types.MapType historicalMap = historicalType.asMapType();
+ Preconditions.checkState(currentMap.keyId() ==
historicalMap.keyId()
+ && currentMap.valueId() ==
historicalMap.valueId(),
+ "Iceberg equality-delete map field ids changed from
(%s, %s) to (%s, %s)",
+ historicalMap.keyId(), historicalMap.valueId(),
+ currentMap.keyId(), currentMap.valueId());
+ Type mergedKey = mergeHistoricalEqualityType(
+ currentMap.keyType(), historicalMap.keyType());
+ Type mergedValue = mergeHistoricalEqualityType(
+ currentMap.valueType(), historicalMap.valueType());
+ if (mergedKey == currentMap.keyType()
+ && mergedValue == currentMap.valueType()) {
+ return currentType;
+ }
+ return currentMap.isValueOptional()
+ ? Types.MapType.ofOptional(
+ currentMap.keyId(), currentMap.valueId(),
+ mergedKey, mergedValue)
+ : Types.MapType.ofRequired(
+ currentMap.keyId(), currentMap.valueId(),
+ mergedKey, mergedValue);
+ default:
+ Preconditions.checkState(currentType.equals(historicalType),
+ "Iceberg equality-delete field type changed from %s to
%s",
+ historicalType, currentType);
+ return currentType;
+ }
+ }
+
+ @VisibleForTesting
+ static boolean requiresRecursiveInitialDefaultMaterialization(
+ Schema scanSchema, List<SlotDescriptor> projectedSlots) {
+ return requiresProjectedIcebergField(scanSchema, projectedSlots,
+ (field, isTopLevel) -> field.initialDefault() != null
+ && (!isTopLevel || field.type().isNestedType()));
+ }
+
+ @VisibleForTesting
+ static boolean requiresMissingRequiredFieldRejection(
+ Schema scanSchema, List<SlotDescriptor> projectedSlots,
+ Optional<List<Schema>> historicalSchemas) {
+ return !historicalSchemas.isPresent()
+ || requiresMissingRequiredFieldRejection(
+ scanSchema, projectedSlots, historicalSchemas.get());
+ }
+
+ @VisibleForTesting
+ static boolean requiresMissingRequiredFieldRejection(
+ Schema scanSchema, List<SlotDescriptor> projectedSlots,
+ List<Schema> historicalSchemas) {
+ Map<Integer, NestedField> fieldById =
TypeUtil.indexById(scanSchema.asStruct());
+ Map<Integer, Integer> parentById =
TypeUtil.indexParents(scanSchema.asStruct());
+ Set<Integer> collectionWrapperFieldIds = new HashSet<>();
+ collectCollectionWrapperFieldIds(scanSchema.asStruct(),
collectionWrapperFieldIds);
+ Set<Integer> potentiallyMissingRequiredFieldIds = new HashSet<>();
+ for (Schema historicalSchema : historicalSchemas) {
+ Map<Integer, NestedField> historicalFieldById =
+ TypeUtil.indexById(historicalSchema.asStruct());
+ for (NestedField field : fieldById.values()) {
+ NestedField historicalField =
historicalFieldById.get(field.fieldId());
+ if (historicalField != null) {
+ if (!collectionWrapperFieldIds.contains(field.fieldId())
Review Comment:
[P2] Enforce same-ID optional-to-required evolution
This gate misses two present-field transitions: `collectionWrapperFieldIds`
contains every list element/map value stable ID, and `initialDefault() != null`
excludes scalar or struct fields. Neither exception makes a historical explicit
NULL valid: defaults fill missing fields only, and a replaced wrapper is
different from a stable-ID evolution. Both readers keep these carriers nullable
and consult requiredness only when a field is missing, so current backends can
return NULL and source backends are left unfenced. Treat every same-ID
optional-to-required transition as requiring current semantics and validate
present null maps.
##########
be/src/format/transformer/vorc_transformer.cpp:
##########
@@ -319,6 +357,30 @@ std::unique_ptr<orc::Type>
VOrcTransformer::_build_orc_type(
}
}
if (nested_field != nullptr) {
+ const PrimitiveType primitive_type = data_type->get_primitive_type();
+ const auto use_iceberg_binary_type = [&](std::string_view binary_type)
{
+ DORIS_CHECK(is_string_type(primitive_type) ||
is_varbinary(primitive_type) ||
+ primitive_type == TYPE_BINARY);
+ type = orc::createPrimitiveType(orc::BINARY);
Review Comment:
[P1] Fence binary ORC writes from old backends
These Iceberg-specific ORC types and the UUID normalization below exist only
on upgraded BEs, but FE write validation still lets a query-available
smooth-upgrade-source BE execute this sink. The baseline BE receives the same
Iceberg schema yet emits generic STRING/CHAR or unannotated BINARY carriers and
does not convert UUID text to 16 bytes. A distributed insert can therefore
commit different, non-Iceberg physical schemas depending on which BE writes
each file. Add a write capability gate or scheduler restriction analogous to
`validateVariantWriteBackendCompatibility()` for ORC schemas containing
UUID/FIXED/BINARY, or provide a wire-compatible fallback.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergWriteSchemaContext.java:
##########
@@ -0,0 +1,842 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.iceberg;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.datasource.mvcc.MvccSnapshot;
+import org.apache.doris.datasource.mvcc.MvccUtil;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Array;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.CreateMap;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.CreateNamedStruct;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Unhex;
+import org.apache.doris.nereids.trees.expressions.literal.ArrayLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.DateTimeV2Literal;
+import org.apache.doris.nereids.trees.expressions.literal.DateV2Literal;
+import org.apache.doris.nereids.trees.expressions.literal.DecimalV3Literal;
+import org.apache.doris.nereids.trees.expressions.literal.DoubleLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.FloatLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.literal.MapLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.StructLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.TimestampTzLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.VarBinaryLiteral;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.DateTimeV2Type;
+import org.apache.doris.nereids.types.DecimalV3Type;
+import org.apache.doris.nereids.types.StructType;
+import org.apache.doris.nereids.types.TimeStampTzType;
+import org.apache.doris.nereids.types.VarBinaryType;
+import org.apache.doris.nereids.util.TypeCoercionUtils;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.io.BaseEncoding;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionSpecParser;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.SortOrderParser;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.SnapshotUtil;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.UUID;
+
+/**
+ * Statement-scoped Iceberg write schema and write-default values.
+ *
+ * <p>The context pins one Iceberg schema before analysis. The analyzer,
planner sink and
+ * transaction preflight must all use this same instance so a concurrent
schema change cannot
+ * combine expressions from one schema with a writer schema from another one.
+ */
+public final class IcebergWriteSchemaContext {
+ private final long tableId;
+ private final String tableName;
+ private final Schema schema;
+ private final int formatVersion;
+ private final Optional<String> branchName;
+ private final Optional<UUID> tableUuid;
+ private final Optional<String> v1MetadataFileLocation;
+ private final Optional<Long> v1MetadataTimestampMillis;
+ private final String schemaJson;
+ private final Schema mergeSchema;
+ private final String mergeSchemaJson;
+ private final PartitionSpec partitionSpec;
+ private final String partitionSpecJson;
+ private final SortOrder sortOrder;
+ private final String sortOrderJson;
+ private final FileFormat fileFormat;
+ private final MetricsConfig metricsConfig;
+ private final String fileCompression;
+ private final String dataLocation;
+ private final Map<String, String> writerProperties;
+ private final List<Column> columns;
+ private final List<Column> mergeColumns;
+ private final Map<Integer, Types.NestedField> fieldsById;
+ private final Map<Integer, Expression> writeDefaultsById;
+
+ /** Pin the statement snapshot's current table schema under the catalog
authentication boundary. */
+ public static IcebergWriteSchemaContext create(
+ IcebergExternalTable dorisTable, Optional<String> branchName) {
+ Objects.requireNonNull(dorisTable, "dorisTable should not be null");
+ Objects.requireNonNull(branchName, "branchName should not be null");
+ try {
+ return
dorisTable.getCatalog().getExecutionAuthenticator().execute(() -> {
+ Table table = dorisTable.getIcebergTable();
+ Schema schema = branchName.isPresent()
+ ? resolveBranchSchema(table, branchName.get(),
dorisTable.getName())
+ : resolveStatementSchema(table, dorisTable);
+ if (branchName.isPresent()) {
+ validateBranchWriterSchema(
+ schema, table.schema(), branchName.get(),
dorisTable.getName());
+ }
+ int formatVersion = IcebergUtils.getFormatVersion(table);
+ TableIdentity tableIdentity = pinTableIdentity(table,
formatVersion);
+ Map<String, String> properties =
ImmutableMap.copyOf(table.properties());
+ return new IcebergWriteSchemaContext(
+ dorisTable.getId(), dorisTable.getName(), schema,
formatVersion, branchName,
+ tableIdentity.uuid,
tableIdentity.v1MetadataFileLocation,
+ tableIdentity.v1MetadataTimestampMillis,
+ bindPartitionSpec(table.spec(), schema,
dorisTable.getName()),
+ bindSortOrder(table.sortOrder(), schema,
dorisTable.getName()),
+ IcebergUtils.getFileFormat(table),
MetricsConfig.forTable(table),
+ IcebergUtils.getFileCompress(table),
IcebergUtils.dataLocation(table), properties,
+ dorisTable.getCatalog().getEnableMappingVarbinary(),
+ dorisTable.getCatalog().getEnableMappingTimestampTz());
+ });
+ } catch (Exception e) {
+ throw new AnalysisException("Failed to pin Iceberg write schema
for table "
+ + dorisTable.getName() + ": " + e.getMessage(), e);
+ }
+ }
+
+ @VisibleForTesting
+ public static IcebergWriteSchemaContext forSchema(Schema schema, int
formatVersion,
+ boolean enableMappingVarbinary, boolean enableMappingTimestampTz) {
+ return new IcebergWriteSchemaContext(-1L, "test_table", schema,
formatVersion,
+ Optional.empty(), Optional.empty(), Optional.empty(),
Optional.empty(),
+ PartitionSpec.unpartitioned(), SortOrder.unsorted(),
+ FileFormat.PARQUET, MetricsConfig.getDefault(),
+ TableProperties.PARQUET_COMPRESSION_DEFAULT_SINCE_1_4_0,
+ "file:///tmp/test_table/data",
+ ImmutableMap.of(TableProperties.FORMAT_VERSION,
Integer.toString(formatVersion)),
+ enableMappingVarbinary, enableMappingTimestampTz);
+ }
+
+ @VisibleForTesting
+ public static IcebergWriteSchemaContext forSchema(Schema schema, int
formatVersion,
+ PartitionSpec partitionSpec, SortOrder sortOrder, FileFormat
fileFormat,
+ MetricsConfig metricsConfig, String fileCompression, String
dataLocation,
+ Map<String, String> writerProperties,
+ boolean enableMappingVarbinary, boolean enableMappingTimestampTz) {
+ return new IcebergWriteSchemaContext(-1L, "test_table", schema,
formatVersion,
+ Optional.empty(), Optional.empty(), Optional.empty(),
Optional.empty(),
+ partitionSpec, sortOrder, fileFormat, metricsConfig,
+ fileCompression, dataLocation, writerProperties,
+ enableMappingVarbinary, enableMappingTimestampTz);
+ }
+
+ @VisibleForTesting
+ static IcebergWriteSchemaContext forSchemaWithUuidIdentity(
+ Schema schema, int formatVersion, UUID tableUuid) {
+ return new IcebergWriteSchemaContext(
+ -1L, "test_table", schema, formatVersion, Optional.empty(),
+ Optional.of(tableUuid), Optional.empty(), Optional.empty(),
+ PartitionSpec.unpartitioned(), SortOrder.unsorted(),
FileFormat.PARQUET,
+ MetricsConfig.getDefault(),
+ TableProperties.PARQUET_COMPRESSION_DEFAULT_SINCE_1_4_0,
+ "file:///tmp/test_table/data",
+ ImmutableMap.of(TableProperties.FORMAT_VERSION,
Integer.toString(formatVersion)),
+ true, true);
+ }
+
+ private IcebergWriteSchemaContext(long tableId, String tableName, Schema
schema,
+ int formatVersion, Optional<String> branchName, Optional<UUID>
tableUuid,
+ Optional<String> v1MetadataFileLocation,
+ Optional<Long> v1MetadataTimestampMillis,
+ PartitionSpec partitionSpec, SortOrder sortOrder, FileFormat
fileFormat,
+ MetricsConfig metricsConfig, String fileCompression, String
dataLocation,
+ Map<String, String> writerProperties,
+ boolean enableMappingVarbinary, boolean enableMappingTimestampTz) {
+ this.tableId = tableId;
+ this.tableName = Objects.requireNonNull(tableName, "tableName should
not be null");
+ this.schema = Objects.requireNonNull(schema, "schema should not be
null");
+ this.formatVersion = formatVersion;
+ this.branchName = Objects.requireNonNull(branchName, "branchName
should not be null");
+ this.tableUuid = Objects.requireNonNull(tableUuid, "tableUuid should
not be null");
+ this.v1MetadataFileLocation = Objects.requireNonNull(
+ v1MetadataFileLocation, "v1MetadataFileLocation should not be
null");
+ this.v1MetadataTimestampMillis = Objects.requireNonNull(
+ v1MetadataTimestampMillis, "v1MetadataTimestampMillis should
not be null");
+ Preconditions.checkState(
+ this.v1MetadataFileLocation.isPresent()
+ == this.v1MetadataTimestampMillis.isPresent(),
+ "Iceberg V1 metadata identity must contain both location and
timestamp");
+ Preconditions.checkState(
+ !this.tableUuid.isPresent() ||
!this.v1MetadataFileLocation.isPresent(),
+ "Iceberg table identity cannot contain both UUID and V1
metadata");
+ this.schemaJson = SchemaParser.toJson(schema);
+ this.mergeSchema = formatVersion >=
IcebergUtils.ICEBERG_ROW_LINEAGE_MIN_VERSION
+ ? IcebergUtils.appendRowLineageFieldsForV3(schema) : schema;
+ this.mergeSchemaJson = SchemaParser.toJson(mergeSchema);
+ this.partitionSpec = Objects.requireNonNull(partitionSpec,
"partitionSpec should not be null");
+ this.partitionSpecJson = PartitionSpecParser.toJson(partitionSpec);
+ this.sortOrder = Objects.requireNonNull(sortOrder, "sortOrder should
not be null");
+ this.sortOrderJson = SortOrderParser.toJson(sortOrder);
+ this.fileFormat = Objects.requireNonNull(fileFormat, "fileFormat
should not be null");
+ this.metricsConfig = Objects.requireNonNull(metricsConfig,
"metricsConfig should not be null");
+ this.fileCompression = Objects.requireNonNull(
+ fileCompression, "fileCompression should not be null");
+ this.dataLocation = Objects.requireNonNull(dataLocation, "dataLocation
should not be null");
+ this.writerProperties = ImmutableMap.copyOf(
+ Objects.requireNonNull(writerProperties, "writerProperties
should not be null"));
+ validateWriterMetadataSources(schema, partitionSpec, sortOrder,
tableName);
+
+ List<Column> parsedColumns = IcebergUtils.parseSchema(
+ schema, enableMappingVarbinary, enableMappingTimestampTz);
+ this.columns = ImmutableList.copyOf(parsedColumns);
+ List<Column> writerColumns = new ArrayList<>(parsedColumns);
+ writerColumns.add(IcebergRowId.createHiddenColumn());
+ if (formatVersion >= IcebergUtils.ICEBERG_ROW_LINEAGE_MIN_VERSION) {
+ Column rowIdColumn = IcebergUtils.parseField(
+ org.apache.iceberg.MetadataColumns.ROW_ID,
+ enableMappingVarbinary, enableMappingTimestampTz);
+ rowIdColumn.setIsVisible(false);
+ writerColumns.add(rowIdColumn);
+ Column sequenceColumn = IcebergUtils.parseField(
+
org.apache.iceberg.MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER,
+ enableMappingVarbinary, enableMappingTimestampTz);
+ sequenceColumn.setIsVisible(false);
+ writerColumns.add(sequenceColumn);
+ }
+ this.mergeColumns = ImmutableList.copyOf(writerColumns);
+
+ ImmutableMap.Builder<Integer, Types.NestedField> byId =
ImmutableMap.builder();
+ ImmutableMap.Builder<Integer, Expression> defaults =
ImmutableMap.builder();
+ for (Types.NestedField field : schema.columns()) {
+ byId.put(field.fieldId(), field);
+ if (field.writeDefault() != null) {
+ DataType targetType =
DataType.fromCatalogType(IcebergUtils.icebergTypeToDorisType(
+ field.type(), enableMappingVarbinary,
enableMappingTimestampTz));
+ defaults.put(field.fieldId(), toDorisExpression(
+ field.type(), field.writeDefault(), targetType,
+ enableMappingVarbinary, enableMappingTimestampTz));
+ }
+ }
+ this.fieldsById = byId.build();
+ this.writeDefaultsById = defaults.build();
+ }
+
+ private static PartitionSpec bindPartitionSpec(
+ PartitionSpec partitionSpec, Schema schema, String tableName) {
+ if (!partitionSpec.isPartitioned()) {
+ return PartitionSpec.builderFor(schema)
+ .withSpecId(partitionSpec.specId())
+ .build();
+ }
+ try {
+ return PartitionSpecParser.fromJson(schema,
PartitionSpecParser.toJson(partitionSpec));
+ } catch (RuntimeException e) {
+ throw new AnalysisException("Iceberg partition spec " +
partitionSpec.specId()
+ + " is incompatible with pinned schema " +
schema.schemaId()
+ + " for table " + tableName + ": " + e.getMessage(), e);
+ }
+ }
+
+ private static SortOrder bindSortOrder(SortOrder sortOrder, Schema schema,
String tableName) {
+ if (!sortOrder.isSorted()) {
+ return SortOrder.unsorted();
+ }
+ try {
+ return SortOrderParser.fromJson(schema,
SortOrderParser.toJson(sortOrder));
+ } catch (RuntimeException e) {
+ throw new AnalysisException("Iceberg sort order " +
sortOrder.orderId()
+ + " is incompatible with pinned schema " +
schema.schemaId()
+ + " for table " + tableName + ": " + e.getMessage(), e);
+ }
+ }
+
+ private static void validateWriterMetadataSources(
+ Schema schema, PartitionSpec partitionSpec, SortOrder sortOrder,
String tableName) {
+ Map<Integer, Types.NestedField> topLevelFields =
schema.columns().stream()
+
.collect(ImmutableMap.toImmutableMap(Types.NestedField::fieldId, field ->
field));
+ for (PartitionField field : partitionSpec.fields()) {
+ if (!topLevelFields.containsKey(field.sourceId())) {
+ throw new AnalysisException("Iceberg partition field " +
field.fieldId()
+ + " references source field " + field.sourceId()
+ + " outside pinned top-level schema " +
schema.schemaId()
+ + " for table " + tableName);
+ }
+ }
+ for (SortField field : sortOrder.fields()) {
+ if (schema.findField(field.sourceId()) == null) {
+ throw new AnalysisException("Iceberg sort field references
source field "
+ + field.sourceId() + " outside pinned schema " +
schema.schemaId()
+ + " for table " + tableName);
+ }
+ }
+ }
+
+ private static Schema resolveBranchSchema(Table table, String branchName,
String tableName) {
+ SnapshotRef ref = table.refs().get(branchName);
+ if (ref == null) {
+ throw new AnalysisException(branchName + " is not founded in " +
tableName);
+ }
+ if (!ref.isBranch()) {
+ throw new AnalysisException(branchName
+ + " is a tag, not a branch. Tags cannot be targets for
producing snapshots");
+ }
+ return SnapshotUtil.schemaFor(table, ref.snapshotId());
+ }
+
+ private static Schema resolveStatementSchema(Table table,
IcebergExternalTable dorisTable) {
+ Optional<MvccSnapshot> snapshot =
MvccUtil.getSnapshotFromContext(dorisTable);
+ if (!snapshot.isPresent()) {
+ return table.schema();
+ }
+ Preconditions.checkState(snapshot.get() instanceof IcebergMvccSnapshot,
+ "Expected an Iceberg MVCC snapshot for table %s",
dorisTable.getName());
+ long schemaId = ((IcebergMvccSnapshot) snapshot.get())
+ .getSnapshotCacheValue().getSnapshot().getSchemaId();
+ Schema schema = table.schemas().get(Math.toIntExact(schemaId));
+ return Preconditions.checkNotNull(schema,
+ "Iceberg schema %s is not available in the statement table
metadata for %s",
+ schemaId, dorisTable.getName());
+ }
+
+ /**
+ * Reject branch writes whose files cannot satisfy the table-current
schema.
+ *
+ * <p>Iceberg resolves columns from the branch-head schema, but stamps the
new branch snapshot
+ * with the table-current schema. A field that is present in both schemas
must remain required
+ * because an optional branch writer can emit an explicit null that no
initial default repairs.
+ * A field absent from the branch can rely on an initial default.
+ */
+ private static void validateBranchWriterSchema(
+ Schema branchSchema, Schema currentSchema, String branchName,
String tableName) {
+ Map<Integer, Types.NestedField> branchFields =
+ TypeUtil.indexById(branchSchema.asStruct());
+ Map<Integer, Types.NestedField> currentFields =
+ TypeUtil.indexById(currentSchema.asStruct());
+ Map<Integer, Integer> currentParents =
+ TypeUtil.indexParents(currentSchema.asStruct());
+ for (Types.NestedField currentField : currentFields.values()) {
+ Types.NestedField branchField =
branchFields.get(currentField.fieldId());
+ if (branchField != null) {
+ if (currentField.isRequired() && branchField.isOptional()) {
+ throw incompatibleBranchSchema(
+ branchSchema, currentSchema, branchName,
tableName, currentField,
+ "is optional in the pinned branch schema and can
contain explicit nulls");
+ }
+ continue;
+ }
+ Types.NestedField highestMissingField = currentField;
+ Integer parentId = currentParents.get(currentField.fieldId());
+ while (parentId != null && !branchFields.containsKey(parentId)) {
+ highestMissingField =
Preconditions.checkNotNull(currentFields.get(parentId),
+ "Iceberg parent field %s is absent from current
schema", parentId);
+ parentId = currentParents.get(parentId);
+ }
+ if (highestMissingField.isRequired()
+ && highestMissingField.initialDefault() == null) {
+ throw incompatibleBranchSchema(
+ branchSchema, currentSchema, branchName, tableName,
highestMissingField,
+ "is absent from the pinned branch schema and has no
initial default");
+ }
+ }
+ }
+
+ private static AnalysisException incompatibleBranchSchema(
+ Schema branchSchema, Schema currentSchema, String branchName,
String tableName,
+ Types.NestedField field, String incompatibility) {
+ return new AnalysisException("Iceberg table current schema " +
currentSchema.schemaId()
+ + " cannot label files written with pinned branch " +
branchName + " schema "
+ + branchSchema.schemaId() + " for table " + tableName + ":
required field "
+ + field.name() + " (id " + field.fieldId() + ") " +
incompatibility
+ + "; retry after updating the branch schema");
+ }
+
+ /** Resolve a write default by the pinned target field name. */
+ public Expression resolveWriteDefault(String columnName) {
+ Column column = columns.stream()
+ .filter(targetColumn ->
targetColumn.getName().equalsIgnoreCase(columnName))
+ .findFirst()
+ .orElseThrow(() -> new AnalysisException(
+ "Cannot find column information for DEFAULT(" +
columnName + ")"));
+ return resolveWriteDefault(column);
+ }
+
+ /** Resolve the value used for an omitted column or an explicit DEFAULT. */
+ public Expression resolveWriteDefault(Column column) {
+ Types.NestedField field = fieldsById.get(column.getUniqueId());
+ if (field == null) {
+ throw new AnalysisException("Column " + column.getName()
+ + " is not present in pinned Iceberg schema " +
getSchemaId());
+ }
+ Expression writeDefault = writeDefaultsById.get(field.fieldId());
+ if (writeDefault != null) {
+ return writeDefault;
+ }
+ DataType targetType = DataType.fromCatalogType(column.getType());
+ if (field.isOptional()) {
+ return new NullLiteral(targetType);
+ }
+ throw new AnalysisException("Column has no write default and is
required, column=" + field.name());
+ }
+
+ /** Validate that the fresh table can commit files described by the pinned
writer metadata. */
+ public void validateCurrentSchema(Table table) {
+ validateCurrentSchema(table, false);
+ }
+
+ /**
+ * Validate that the fresh table can commit files described by the pinned
writer metadata.
+ *
+ * <p>Every overwrite additionally requires the pinned spec to remain
current because both
+ * dynamic replacement and static replacement semantics depend on whether
and how that spec is
+ * partitioned. Appends can safely write an older retained spec, so they
only require the pinned
+ * definition to remain available.
+ */
+ public void validateCurrentSchema(Table table, boolean
requireCurrentPartitionSpec) {
+ Schema currentSchema = branchName.isPresent()
+ ? resolveBranchSchema(table, branchName.get(), tableName)
+ : table.schema();
+ int currentFormatVersion = IcebergUtils.getFormatVersion(table);
+ validateTableIdentity(table, currentFormatVersion);
+ if (currentSchema.schemaId() != getSchemaId() || currentFormatVersion
!= formatVersion) {
+ throw new AnalysisException("Iceberg table schema changed during
write planning for " + tableName
+ + ": pinned schema " + getSchemaId() + "/format " +
formatVersion
+ + ", current schema " + currentSchema.schemaId() +
"/format " + currentFormatVersion
+ + "; retry the statement");
+ }
+ String currentDataLocation = IcebergUtils.dataLocation(table);
+ if (!dataLocation.equals(currentDataLocation)
+ || !writerProperties.equals(table.properties())) {
+ throw new AnalysisException("Iceberg table writer properties or
data location changed during "
+ + "write planning for " + tableName + "; retry the
statement");
+ }
+ if (branchName.isPresent()) {
+ validateBranchWriterSchema(
+ schema, table.schema(), branchName.get(), tableName);
+ }
+ PartitionSpec currentSpec = table.specs().get(partitionSpec.specId());
+ if (currentSpec == null ||
!partitionSpecJson.equals(PartitionSpecParser.toJson(currentSpec))) {
+ throw new AnalysisException("Iceberg partition spec changed during
write planning for "
+ + tableName + ": pinned spec " + partitionSpec.specId()
+ + " is not available with the same definition; retry the
statement");
+ }
+ if (requireCurrentPartitionSpec) {
+ PartitionSpec activeSpec = table.spec();
+ if (activeSpec.specId() != partitionSpec.specId()
+ ||
!partitionSpecJson.equals(PartitionSpecParser.toJson(activeSpec))) {
+ throw new AnalysisException("Iceberg current partition spec
changed during overwrite "
+ + "planning for " + tableName + ": pinned spec " +
partitionSpec.specId()
+ + ", current spec " + activeSpec.specId() + "; retry
the statement");
+ }
+ }
+ SortOrder currentSortOrder =
table.sortOrders().get(sortOrder.orderId());
+ if (currentSortOrder == null ||
!sortOrderJson.equals(SortOrderParser.toJson(currentSortOrder))) {
+ throw new AnalysisException("Iceberg sort order changed during
write planning for "
+ + tableName + ": pinned order " + sortOrder.orderId()
+ + " is not available with the same definition; retry the
statement");
+ }
+ }
+
+ private static TableIdentity pinTableIdentity(Table table, int
formatVersion) {
+ if (table instanceof HasTableOperations) {
+ TableMetadata metadata = Preconditions.checkNotNull(
+ ((HasTableOperations) table).operations().current(),
+ "Iceberg table %s has no current metadata", table.name());
+ if (metadata.uuid() != null) {
+ return TableIdentity.forUuid(UUID.fromString(metadata.uuid()));
+ }
+ Preconditions.checkState(formatVersion == 1,
+ "Iceberg table %s format %s has no table UUID",
table.name(), formatVersion);
+ return TableIdentity.forV1Metadata(
+ Preconditions.checkNotNull(metadata.metadataFileLocation(),
+ "Iceberg V1 table %s has no metadata file
location", table.name()),
+ metadata.lastUpdatedMillis());
+ }
+ return TableIdentity.forUuid(Preconditions.checkNotNull(
+ table.uuid(), "Iceberg table %s does not expose a table UUID",
table.name()));
+ }
+
+ private void validateTableIdentity(Table table, int currentFormatVersion) {
+ if (tableUuid.isPresent()) {
+ TableIdentity currentIdentity = pinTableIdentity(table,
currentFormatVersion);
+ if (!tableUuid.equals(currentIdentity.uuid)) {
+ throw tableIdentityChanged();
+ }
+ return;
+ }
+ if (!v1MetadataFileLocation.isPresent()) {
+ return;
+ }
+ Preconditions.checkState(table instanceof HasTableOperations,
+ "Iceberg V1 table %s does not expose table operations",
table.name());
+ TableMetadata currentMetadata = Preconditions.checkNotNull(
+ ((HasTableOperations) table).operations().current(),
+ "Iceberg V1 table %s has no current metadata", table.name());
+ boolean sameMetadata = v1MetadataFileLocation.get().equals(
+ currentMetadata.metadataFileLocation())
+ && v1MetadataTimestampMillis.get() ==
currentMetadata.lastUpdatedMillis();
+ boolean retainedAncestor = currentMetadata.previousFiles().stream()
+ .anyMatch(entry ->
v1MetadataFileLocation.get().equals(entry.file())
+ && v1MetadataTimestampMillis.get() ==
entry.timestampMillis());
+ if (!sameMetadata && !retainedAncestor) {
+ throw tableIdentityChanged();
+ }
+ }
+
+ private AnalysisException tableIdentityChanged() {
+ return new AnalysisException("Iceberg table identity changed during
write planning for "
+ + tableName + "; the table may have been dropped and
recreated; retry the statement");
+ }
+
+ private static final class TableIdentity {
+ private final Optional<UUID> uuid;
+ private final Optional<String> v1MetadataFileLocation;
+ private final Optional<Long> v1MetadataTimestampMillis;
+
+ private TableIdentity(Optional<UUID> uuid, Optional<String>
v1MetadataFileLocation,
+ Optional<Long> v1MetadataTimestampMillis) {
+ this.uuid = uuid;
+ this.v1MetadataFileLocation = v1MetadataFileLocation;
+ this.v1MetadataTimestampMillis = v1MetadataTimestampMillis;
+ }
+
+ private static TableIdentity forUuid(UUID uuid) {
+ return new TableIdentity(
+ Optional.of(uuid), Optional.empty(), Optional.empty());
+ }
+
+ private static TableIdentity forV1Metadata(
+ String metadataFileLocation, long metadataTimestampMillis) {
+ return new TableIdentity(
+ Optional.empty(), Optional.of(metadataFileLocation),
+ Optional.of(metadataTimestampMillis));
+ }
+ }
+
+ public int getSchemaId() {
+ return schema.schemaId();
+ }
+
+ public int getFormatVersion() {
+ return formatVersion;
+ }
+
+ public Optional<String> getBranchName() {
+ return branchName;
+ }
+
+ public boolean isTargetTable(long candidateTableId) {
+ return tableId == candidateTableId;
+ }
+
+ public String getSchemaJson() {
+ return schemaJson;
+ }
+
+ public String getMergeSchemaJson() {
+ return mergeSchemaJson;
+ }
+
+ public Schema getMergeSchema() {
+ return mergeSchema;
+ }
+
+ public Schema getSchema() {
+ return schema;
+ }
+
+ public PartitionSpec getPartitionSpec() {
+ return partitionSpec;
+ }
+
+ public String getPartitionSpecJson() {
+ return partitionSpecJson;
+ }
+
+ public SortOrder getSortOrder() {
+ return sortOrder;
+ }
+
+ public FileFormat getFileFormat() {
+ return fileFormat;
+ }
+
+ public MetricsConfig getMetricsConfig() {
+ return metricsConfig;
+ }
+
+ public String getFileCompression() {
+ return fileCompression;
+ }
+
+ public String getDataLocation() {
+ return dataLocation;
+ }
+
+ public List<Column> getColumns() {
+ return columns;
+ }
+
+ public List<Column> getMergeColumns() {
+ return mergeColumns;
+ }
+
+ public Optional<Types.NestedField> findField(Column column) {
+ return Optional.ofNullable(fieldsById.get(column.getUniqueId()));
+ }
+
+ @VisibleForTesting
+ static Expression toDorisExpression(Type icebergType, Object value,
DataType targetType,
+ boolean enableMappingVarbinary, boolean enableMappingTimestampTz) {
+ Objects.requireNonNull(icebergType, "icebergType should not be null");
+ Objects.requireNonNull(targetType, "targetType should not be null");
+ if (value == null) {
+ return new NullLiteral(targetType);
+ }
+
+ switch (icebergType.typeId()) {
+ case BOOLEAN:
+ return BooleanLiteral.of((Boolean) value);
+ case INTEGER:
+ return new IntegerLiteral((Integer) value);
+ case LONG:
+ return new BigIntLiteral((Long) value);
+ case FLOAT:
+ return new FloatLiteral((Float) value);
+ case DOUBLE:
+ return new DoubleLiteral((Double) value);
+ case DECIMAL:
+ return new DecimalV3Literal((DecimalV3Type) targetType,
(BigDecimal) value);
+ case STRING:
+ return new StringLiteral((String) value);
+ case UUID:
+ return binaryExpression(uuidBytes((UUID) value), targetType);
+ case FIXED:
+ case BINARY:
+ return binaryExpression(byteBufferBytes((ByteBuffer) value),
targetType);
+ case DATE:
+ LocalDate date = LocalDate.ofEpochDay(((Integer)
value).longValue());
+ return new DateV2Literal(date.getYear(), date.getMonthValue(),
date.getDayOfMonth());
+ case TIMESTAMP:
+ long micros = (Long) value;
+ LocalDateTime dateTime = microsToDateTime(micros);
+ long microsecond = Math.floorMod(micros, 1_000_000L);
+ Types.TimestampType timestampType = (Types.TimestampType)
icebergType;
+ if (enableMappingTimestampTz &&
timestampType.shouldAdjustToUTC()) {
+ return new TimestampTzLiteral((TimeStampTzType) targetType,
+ dateTime.getYear(), dateTime.getMonthValue(),
+ dateTime.getDayOfMonth(), dateTime.getHour(),
dateTime.getMinute(),
+ dateTime.getSecond(), microsecond);
+ }
+ return new DateTimeV2Literal((DateTimeV2Type) targetType,
Review Comment:
[P1] Keep legacy timestamptz defaults as instants
With timestamp-tz mapping disabled, this turns the Iceberg default's UTC
microsecond instant into a timezone-less UTC-wall `DATETIMEV2`. Parquet's
Iceberg Arrow schema and the ORC serializer then interpret that wall time in
the session timezone, so in Asia/Shanghai a `01:02Z` default is persisted as
`17:02Z` on the prior day. Missing-field defaults have the sibling
inconsistency: their offset is stripped while physical adjusted-to-UTC values
decode to session-local time. The new UTC-only regression masks both paths.
Convert defaults to the session-local wall time consistently, or retain an
instant-aware carrier through serialization.
--
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]
