github-actions[bot] commented on code in PR #69032:
URL: https://github.com/apache/doris/pull/69032#discussion_r4294126310
##########
be/src/exec/common/hash_table/hash_key_type.h:
##########
@@ -125,7 +125,11 @@ inline HashKeyType get_hash_key_type(const
std::vector<DataTypePtr>& data_types)
// fixed-width or string-key specialization even when it is the only
grouping/join key.
return HashKeyType::serialized;
}
- // serialized cannot be used in the case of single column, because the
join operator will have some processing of column nullable, resulting in
incorrect serialized results.
+ // Spatial columns store StringViews, not STRING offsets/chars. Use their
column serializer
+ // for both values and nullable markers, as for the existing single-column
complex keys.
+ if (t->get_primitive_type() == TYPE_GEOMETRY || t->get_primitive_type() ==
TYPE_GEOGRAPHY) {
Review Comment:
[P2] Handle spatial predicate dispatch for actual scan columns
For `HashJoin(l.geom = r.geom)` over actual Iceberg scans, the newly
admitted spatial key can generate an IN/Bloom runtime filter:
`RuntimeFilterPushDownVisitor` accepts the bare spatial slot, but
`RuntimeFilter::_init_with_desc -> RuntimeFilterWrapper ->
create_predicate_function` has no GEOMETRY/GEOGRAPHY case and throws `predicate
with type GEOMETRY` (likewise GEOGRAPHY) during setup.
`SET_VAR(enable_runtime_filter_prune=false, runtime_filter_type=1)` makes this
deterministic. The constructor-over-numbers join regression avoids that path
because nonnumeric function probes reject runtime filters. Constant-list
spatial IN/NOT IN also reaches the same unsupported factory. Implement matching
spatial predicate storage, or suppress these runtime filters and reject
unsupported IN during analysis, and cover joins of real Iceberg spatial
columns. This is distinct from the repaired serialized hash-key dispatch.
##########
be/src/format/transformer/viceberg_parquet_writer.cpp:
##########
@@ -18,15 +18,128 @@
#include "format/transformer/viceberg_parquet_writer.h"
#include <parquet/api/reader.h>
+#include <parquet/arrow/schema.h>
+#include <parquet/file_writer.h>
#include <parquet/schema.h>
#include <algorithm>
+#include <numeric>
+#include <string>
+#include "exprs/function/geo/functions_geo.h"
#include "format/table/iceberg/iceberg_arrow_block_convertor.h"
#include "format/table/parquet_utils.h"
+#include "runtime/exec_env.h"
#include "runtime/runtime_state.h"
namespace doris {
+namespace {
+
+::parquet::schema::NodePtr clone_parquet_node(const ::parquet::schema::Node&
node) {
+ if (node.is_primitive()) {
+ const auto& primitive = static_cast<const
::parquet::schema::PrimitiveNode&>(node);
+ return ::parquet::schema::PrimitiveNode::Make(
+ primitive.name(), primitive.repetition(),
primitive.logical_type(),
+ primitive.physical_type(), primitive.type_length(),
primitive.field_id());
+ }
+
+ const auto& group = static_cast<const ::parquet::schema::GroupNode&>(node);
+ ::parquet::schema::NodeVector fields;
+ fields.reserve(group.field_count());
+ for (int index = 0; index < group.field_count(); ++index) {
+ fields.push_back(clone_parquet_node(*group.field(index)));
+ }
+ return ::parquet::schema::GroupNode::Make(
+ group.name(), group.repetition(), fields,
+ group.logical_type()->is_none() ? nullptr : group.logical_type(),
group.field_id());
+}
+
+arrow::Result<::parquet::LogicalType::EdgeInterpolationAlgorithm>
geography_algorithm(
+ const std::string& algorithm) {
+ using Algorithm = ::parquet::LogicalType::EdgeInterpolationAlgorithm;
+ if (algorithm == "spherical") {
+ return Algorithm::SPHERICAL;
+ }
+ if (algorithm == "vincenty") {
+ return Algorithm::VINCENTY;
+ }
+ if (algorithm == "thomas") {
+ return Algorithm::THOMAS;
+ }
+ if (algorithm == "andoyer") {
+ return Algorithm::ANDOYER;
+ }
+ if (algorithm == "karney") {
+ return Algorithm::KARNEY;
+ }
+ return arrow::Status::Invalid("Unsupported Iceberg geography edge
algorithm: ", algorithm);
+}
+
+bool contains_spatial_type(iceberg::Type* type) {
+ switch (type->type_id()) {
+ case iceberg::TypeID::GEOMETRY:
+ case iceberg::TypeID::GEOGRAPHY:
+ return true;
+ case iceberg::TypeID::STRUCT:
+ for (const auto& field : type->as_struct_type()->fields()) {
+ if (contains_spatial_type(field.field_type())) {
+ return true;
+ }
+ }
+ return false;
+ case iceberg::TypeID::LIST:
+ return
contains_spatial_type(type->as_list_type()->element_field().field_type());
+ case iceberg::TypeID::MAP:
+ return
contains_spatial_type(type->as_map_type()->key_field().field_type()) ||
+
contains_spatial_type(type->as_map_type()->value_field().field_type());
+ default:
+ return false;
+ }
+}
+
+arrow::Result<std::shared_ptr<::parquet::schema::GroupNode>>
make_iceberg_parquet_schema(
+ const std::shared_ptr<arrow::Schema>& arrow_schema, const
iceberg::Schema& iceberg_schema,
+ const std::shared_ptr<::parquet::WriterProperties>& writer_properties,
+ const std::shared_ptr<::parquet::ArrowWriterProperties>&
arrow_properties) {
+ std::shared_ptr<::parquet::SchemaDescriptor> parquet_schema;
+ ARROW_RETURN_NOT_OK(::parquet::arrow::ToParquetSchema(arrow_schema.get(),
*writer_properties,
+ *arrow_properties,
&parquet_schema));
+
+ const auto* source_root = parquet_schema->group_node();
+ ::parquet::schema::NodeVector fields;
+ fields.reserve(source_root->field_count());
+ for (int index = 0; index < source_root->field_count(); ++index) {
+ const auto& source_field = source_root->field(index);
+ const auto& iceberg_field = iceberg_schema.columns()[index];
+ if (iceberg_field.field_type()->type_id() ==
iceberg::TypeID::GEOMETRY) {
+ const auto& geometry =
+ static_cast<const
iceberg::GeometryType&>(*iceberg_field.field_type());
+ fields.push_back(::parquet::schema::PrimitiveNode::Make(
+ source_field->name(), source_field->repetition(),
+ ::parquet::LogicalType::Geometry(geometry.crs()),
::parquet::Type::BYTE_ARRAY,
Review Comment:
[P2] Preserve referenced CRS definitions in Parquet file metadata
A valid Iceberg `geometry(projjson:my_crs)` or Geography field resolves
`my_crs` from a [table property](https://iceberg.apache.org/spec/#crs), but
[Parquet resolves that reference in file key-value
metadata](https://github.com/apache/parquet-format/blob/master/Geospatial.md#coordinate-reference-system).
Copying the CRS string here does not copy its definition: the writer receives
only schema/schema JSON, and the Arrow schema carries no referenced property.
Moreover, `_open_file_writer` omits the metadata argument to
`ParquetFileWriter::Open`; pinned Arrow 24 `FileWriter::Make` does not
serialize schema metadata. INSERT SELECT between matching custom-CRS tables
therefore commits files whose declared CRS cannot be resolved by a CRS-aware
Parquet reader. Carry the referenced definitions into file metadata, or reject
such writes before commit, and cover both spatial kinds. Adding the definitions
only to the Arrow schema would still lose them on this open path.
--
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]
