Gabriel39 commented on code in PR #68934:
URL: https://github.com/apache/doris/pull/68934#discussion_r4255722210
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java:
##########
@@ -929,14 +930,19 @@ private Expression buildPartitionFilter(Map<String,
String> staticPartitions, Pa
* both paths that build this predicate: DELETE/UPDATE/MERGE conflict
detection and
* {@code INSERT OVERWRITE ... PARTITION(d='nan')}.
*/
- private static Expression identityPartitionPredicate(String sourceColName,
Object value) {
+ private static Expression identityPartitionPredicate(Types.NestedField
sourceField, Object value) {
+ String sourceColName = sourceField.name();
if (value == null) {
return Expressions.isNull(sourceColName);
}
if ((value instanceof Double || value instanceof Float)
&& Double.isNaN(((Number) value).doubleValue())) {
return Expressions.isNaN(sourceColName);
}
+ if (sourceField.type().typeId() == TypeID.TIMESTAMP_NANO) {
+ // Partition values are already nanos; a raw Long would be rebound
as microseconds.
+ return Expressions.equal(sourceColName, Expressions.nanos((Long)
value));
Review Comment:
I will share the table-history safety check with generated partition
predicates. Static partition overwrite must fail before committing when
retained DATE bounds make the nanos predicate unsafe; conflict-only filters
will conservatively omit unsafe predicates. I will cover retained unpartitioned
DATE files followed by timestamp_ns and partition evolution.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPredicateConverter.java:
##########
@@ -475,7 +507,10 @@ private Types.NestedField getPushdownField(String colName)
{
||
ICEBERG_LAST_UPDATED_SEQUENCE_NUMBER_COL.equalsIgnoreCase(colName)) {
return null;
}
- return schema.caseInsensitiveFindField(colName);
+ Types.NestedField field = schema.caseInsensitiveFindField(colName);
+ // Iceberg 1.11 widens retained four-byte DATE bounds to unscaled
longs. Pruning those
+ // epoch-day bounds with nanosecond predicates can discard matching
files in every planner.
+ return field != null && unsafeNanosFields.contains(field.fieldId()) ?
null : field;
Review Comment:
I will make NOT conversion require an exact conversion of its entire
operand. Dropping an unsafe nanos conjunct is valid for a positive scan filter
but becomes incorrect after negation. I will add coverage for NOT over a
partially convertible AND and verify that matching files are retained.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java:
##########
@@ -801,6 +828,11 @@ static boolean isValidRelatedTable(Table table) {
return false;
}
PartitionField partitionField = fields.get(0);
+ if (spec.schema().findType(partitionField.sourceId()).typeId() ==
Type.TypeID.TIMESTAMP_NANO) {
Review Comment:
I will treat a missing historical partition source type as an unsupported
partition projection, rather than dereferencing it. This is an Iceberg metadata
issue (including v1 tables), not the excluded Doris format-v1 reader path. I
will cover removing a partition field and then dropping its source column.
##########
be/src/core/data_type_serde/data_type_datetimev2_serde.cpp:
##########
@@ -1049,7 +1050,11 @@ Status
DataTypeDateTimeV2SerDe::read_column_from_orc(IColumn& column,
if (orc_serde_utils::orc_decode_row_count(view.rows, view.selected_rows)
== 0) {
return Status::OK();
}
- return decode_timestamp_orc_values(column, view, *view.timezone);
+ const auto* timezone = view.file_type->getKind() == ::orc::TIMESTAMP &&
+ view.plain_timestamp_timezone !=
nullptr
+ ? view.plain_timestamp_timezone
Review Comment:
I will investigate and fix the historical plain-TIMESTAMP interpretation on
the Doris side using producer-aware ORC timestamp semantics, without modifying
the SDK. Validation will distinguish Java raw-offset behavior from the C++
historical writer-timezone behavior.
##########
be/src/format_v2/orc/orc_reader.cpp:
##########
@@ -1646,24 +1715,91 @@ Status OrcReader::_select_stripe_ranges_by_statistics()
{
return Status::OK();
}
-void OrcReader::_apply_current_stripe_range() {
- if (!_state->stripe_pruning_applied ||
_state->selected_stripe_ranges.empty()) {
- return;
+Status OrcReader::_prepare_timestamp_stripe_ranges() {
+ if (!_state->match_writer_timezone) {
+ return Status::OK();
+ }
+ std::vector<uint64_t> stripes;
+ if (_state->stripe_ranges_applied) {
+ for (const auto& range : _state->selected_stripe_ranges) {
+ for (auto stripe = range.first_stripe; stripe < range.last_stripe;
++stripe) {
+ stripes.push_back(stripe);
+ }
+ }
+ } else {
+ RETURN_IF_ERROR(_collect_split_stripes(&stripes));
+ }
+ _state->selected_stripe_ranges.clear();
+ _state->stripe_ranges_applied = true;
+ _reader_statistics.read_row_groups = cast_set<int32_t>(stripes.size());
+ if (stripes.empty()) {
+ return Status::OK();
+ }
+ try {
+ uint64_t first_row = 0;
+ size_t selected = 0;
+ for (uint64_t index = 0; index <= stripes.back(); ++index) {
+ const auto stripe = _state->reader->getStripe(index);
+ const auto end_row = first_row + stripe->getNumberOfRows();
+ if (index == stripes[selected]) {
+ // A RowReader has one timezone, whereas writer timezones
belong to stripes.
+ // Read footer timezone metadata only when entering the
selected stripe.
+ _state->selected_stripe_ranges.push_back({
+ .first_stripe = index,
+ .last_stripe = index + 1,
+ .offset = stripe->getOffset(),
+ .length = stripe->getLength(),
+ .first_row = first_row,
+ .end_row = end_row,
+ });
+ ++selected;
+ }
+ first_row = end_row;
+ }
+ } catch (const std::exception& e) {
+ return Status::InternalError("Failed to prepare ORC timestamp stripe
ranges: {}", e.what());
+ }
+ return Status::OK();
+}
+
+Status OrcReader::_apply_current_stripe_range() {
+ if (!_state->stripe_ranges_applied ||
_state->selected_stripe_ranges.empty()) {
+ return Status::OK();
}
DORIS_CHECK(_state->current_stripe_range <
_state->selected_stripe_ranges.size());
const auto& stripe_range =
_state->selected_stripe_ranges[_state->current_stripe_range];
_state->row_reader_options.range(stripe_range.offset, stripe_range.length);
+ if (_state->match_writer_timezone) {
+ try {
+ const auto stripe =
_state->reader->getStripe(stripe_range.first_stripe);
+ const auto& name = stripe->getWriterTimezone();
+ const auto writer_timezone = name.empty() ? "GMT" : name;
+ if (!TimezoneUtils::find_cctz_time_zone(writer_timezone,
_state->writer_timezone_obj)) {
+ return Status::InvalidArgument("Unknown ORC stripe writer
timezone: {}",
+ writer_timezone);
+ }
+ // Matching zones preserves the SDK's UTC carrier until its
negative-fraction
+ // correction is complete. cctz then applies historical offsets to
that original second.
+ _state->row_reader_options.setTimezoneName(writer_timezone);
Review Comment:
I will make metadata MIN/MAX fall back to row decoding where the ORC
timestamp statistics cannot preserve the same historical civil-time semantics
as the reader. I will add aggregate-versus-row coverage; no SDK change is
planned.
##########
be/src/format_v2/orc/orc_reader.cpp:
##########
@@ -1646,24 +1715,91 @@ Status OrcReader::_select_stripe_ranges_by_statistics()
{
return Status::OK();
}
-void OrcReader::_apply_current_stripe_range() {
- if (!_state->stripe_pruning_applied ||
_state->selected_stripe_ranges.empty()) {
- return;
+Status OrcReader::_prepare_timestamp_stripe_ranges() {
+ if (!_state->match_writer_timezone) {
+ return Status::OK();
+ }
+ std::vector<uint64_t> stripes;
+ if (_state->stripe_ranges_applied) {
+ for (const auto& range : _state->selected_stripe_ranges) {
+ for (auto stripe = range.first_stripe; stripe < range.last_stripe;
++stripe) {
+ stripes.push_back(stripe);
+ }
+ }
+ } else {
+ RETURN_IF_ERROR(_collect_split_stripes(&stripes));
+ }
+ _state->selected_stripe_ranges.clear();
+ _state->stripe_ranges_applied = true;
+ _reader_statistics.read_row_groups = cast_set<int32_t>(stripes.size());
+ if (stripes.empty()) {
+ return Status::OK();
+ }
+ try {
+ uint64_t first_row = 0;
+ size_t selected = 0;
+ for (uint64_t index = 0; index <= stripes.back(); ++index) {
+ const auto stripe = _state->reader->getStripe(index);
+ const auto end_row = first_row + stripe->getNumberOfRows();
+ if (index == stripes[selected]) {
+ // A RowReader has one timezone, whereas writer timezones
belong to stripes.
+ // Read footer timezone metadata only when entering the
selected stripe.
+ _state->selected_stripe_ranges.push_back({
+ .first_stripe = index,
+ .last_stripe = index + 1,
Review Comment:
I will coalesce adjacent selected stripes that use the same effective
timestamp reader timezone, while preserving split boundaries, pruning gaps, row
IDs, and timezone changes. This avoids recreating an SDK RowReader for every
stripe without modifying the SDK.
##########
be/src/format_v2/parquet/native_schema_desc.cpp:
##########
@@ -882,6 +882,11 @@ std::pair<DataTypePtr, bool>
NativeFieldDescriptor::convert_to_doris_type(
// sub-second values are silently truncated by the target SerDe.
ans.first = DataTypeFactory::instance().create_data_type(TYPE_TIMEV2,
nullable, 0, scale);
} else if (logicalType.__isset.TIMESTAMP) {
+ if (logicalType.TIMESTAMP.unit.__isset.NANOS &&
!logicalType.TIMESTAMP.isAdjustedToUTC) {
+ // Materialize nanoseconds before any table cast; DATETIMEV2 would
discard the remainder.
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_TIMESTAMP_NS, nullable);
Review Comment:
I will add ColumnTimeStampNs to the native nullable-column dispatch so
filtered nullable nanos pages use the fused decoding path. Tests will check
both decoded values and the fallback counter.
--
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]
