OIiveirra commented on code in PR #69032: URL: https://github.com/apache/doris/pull/69032#discussion_r4551060138
########## regression-test/suites/external_table_p0/iceberg/test_iceberg_spatial_v3.groovy: ########## @@ -0,0 +1,2029 @@ +// 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. + +import com.amazonaws.auth.AWSStaticCredentialsProvider +import com.amazonaws.auth.BasicAWSCredentials +import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration +import com.amazonaws.services.s3.AmazonS3ClientBuilder +import com.amazonaws.services.s3.model.ListObjectsV2Request + +suite("test_iceberg_spatial_v3", "p0,external,iceberg,external_docker,external_docker_iceberg,nonConcurrent") { + String enabled = context.config.otherConfigs.get("enableIcebergTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("Iceberg test is disabled") + return + } + + String catalogName = "test_iceberg_spatial_v3" + String dbName = "iceberg_spatial_v3_db" + String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port") + String minioPort = context.config.otherConfigs.get("iceberg_minio_port") + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + + def executeCommand = { String command, int timeoutSeconds = 300 -> + StringBuilder stdout = new StringBuilder() + StringBuilder stderr = new StringBuilder() + def process = new ProcessBuilder("/bin/bash", "-c", command).start() + process.consumeProcessOutput(stdout, stderr) + process.waitForOrKill(timeoutSeconds * 1000) + assertEquals(0, process.exitValue(), + "Command failed\nstdout:\n${stdout}\nstderr:\n${stderr}") + return stdout.toString() + } + + String dockerCommand = context.config.otherConfigs.get("externalDockerCommand") ?: "docker" + String sparkContainer = context.config.otherConfigs.get("icebergSparkContainer") + if (sparkContainer == null || sparkContainer.isEmpty()) { + String containers = executeCommand( + "${dockerCommand} ps --format '{{.ID}}\t{{.Names}}'", 30) + def matches = [] + containers.readLines().each { String line -> + String containerId = line.split(/\t/, 2)[0] + String probe = "${dockerCommand} exec ${containerId} bash -lc " + + "'test -f /mnt/SUCCESS && command -v spark-sql >/dev/null'" + try { + executeCommand(probe, 30) + matches.add(containerId) + } catch (Throwable ignored) { + // Only the Spark service has the Iceberg API used to create this V3 schema. + } + } + assertEquals(1, matches.size(), "Expected exactly one usable Spark Iceberg container") + sparkContainer = matches[0] + } + + String javaSource = ''' +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.io.OutputStream; +import java.util.HashMap; +import java.util.UUID; +import java.util.Map; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.SupportsNamespaces; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.types.Types; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; + +public class CreateIcebergSpatialV3Table { + private static void appendParquetRow(Table table, MessageType message, Group row) throws Exception { + String localFile = "/tmp/spatial-fixture-" + UUID.randomUUID() + ".parquet"; + try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(new Path(localFile)).withType(message).build()) { + writer.write(row); + } + OutputFile output = table.io().newOutputFile(table.location() + + "/data/spatial-fixture-" + UUID.randomUUID() + ".parquet"); + long size = Files.size(Paths.get(localFile)); + try (OutputStream stream = output.create()) { + Files.copy(Paths.get(localFile), stream); + } + Files.delete(Paths.get(localFile)); + table.newAppend().appendFile(DataFiles.builder(table.spec()) + .withPath(output.location()).withFormat(FileFormat.PARQUET) + .withFileSizeInBytes(size).withRecordCount(1).build()).commit(); + } + + private static String parquetField(Types.NestedField field) { + String head = (field.isRequired() ? "required " : "optional "); + String tail = " = " + field.fieldId(); + switch (field.type().typeId()) { + case INTEGER: return head + "int32 " + field.name() + tail + "; "; + case STRING: return head + "binary " + field.name() + " (UTF8)" + tail + "; "; + case GEOMETRY: + case GEOGRAPHY: return head + "binary " + field.name() + tail + "; "; + case LIST: return head + "group " + field.name() + " (LIST)" + tail + + " { repeated group list { " + parquetField(field.type().asListType().fields().get(0)) + "} } "; + case MAP: return head + "group " + field.name() + " (MAP)" + tail + + " { repeated group key_value { " + parquetField(field.type().asMapType().fields().get(0)) + + parquetField(field.type().asMapType().fields().get(1)) + "} } "; + case STRUCT: + StringBuilder structure = new StringBuilder(head + "group " + field.name() + tail + " { "); + for (Types.NestedField child : field.type().asStructType().fields()) { + structure.append(parquetField(child)); + } + return structure.append("} ").toString(); + default: throw new IllegalArgumentException("Unexpected fixture type: " + field.type()); + } + } + + private static Binary binary(String hex) { + byte[] bytes = new byte[hex.length() / 2]; + for (int i = 0; i < bytes.length; i++) { + bytes[i] = (byte) Integer.parseInt(hex.substring(2 * i, 2 * i + 2), 16); + } + return Binary.fromConstantByteArray(bytes); + } + + private static void createRequiredNestedTargets(Catalog catalog, Namespace namespace, + Map<String, String> properties) { + for (boolean geography : new boolean[] {false, true}) { + org.apache.iceberg.types.Type spatial = geography + ? Types.GeographyType.crs84() : Types.GeometryType.crs84(); + TableIdentifier name = TableIdentifier.of(namespace, + "spatial_required_" + (geography ? "geography" : "geometry")); + if (catalog.tableExists(name)) { + catalog.dropTable(name); + } + Schema schema = new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "payload", Types.StructType.of( + Types.NestedField.optional(3, "items", Types.ListType.ofRequired(4, spatial)))), + Types.NestedField.optional(5, "entries", Types.MapType.ofOptional( + 6, 7, Types.StringType.get(), Types.ListType.ofRequired(8, spatial)))); + catalog.createTable(name, schema, PartitionSpec.unpartitioned(), properties); + } + } + + private static void createNestedSource(Catalog catalog, Namespace namespace, + Map<String, String> properties) throws Exception { + java.util.List<Types.NestedField> fields = new java.util.ArrayList<>(); + fields.add(Types.NestedField.required(1, "id", Types.IntegerType.get())); + int nextId = 2; + for (boolean geography : new boolean[] {false, true}) { + String prefix = geography ? "gg_" : "g_"; + org.apache.iceberg.types.Type geo = geography ? Types.GeographyType.crs84() : Types.GeometryType.crs84(); + fields.add(Types.NestedField.optional(nextId++, prefix + "array", + Types.ListType.ofOptional(nextId++, geo))); + fields.add(Types.NestedField.optional(nextId++, prefix + "map", + Types.MapType.ofOptional(nextId++, nextId++, Types.StringType.get(), geo))); + fields.add(Types.NestedField.optional(nextId++, prefix + "struct", Types.StructType.of( + Types.NestedField.optional(nextId++, "shape", geo), + Types.NestedField.optional(nextId++, "label", Types.StringType.get())))); + fields.add(Types.NestedField.optional(nextId++, prefix + "deep", + Types.MapType.ofOptional(nextId++, nextId++, Types.StringType.get(), + Types.ListType.ofOptional(nextId++, Types.StructType.of( + Types.NestedField.optional(nextId++, "shape", geo)))))); + } + TableIdentifier name = TableIdentifier.of(namespace, "spatial_nested_source"); + if (catalog.tableExists(name)) { + catalog.dropTable(name); + } + Table table = catalog.createTable(name, new Schema(fields), PartitionSpec.unpartitioned(), properties); + StringBuilder schema = new StringBuilder("message nested_spatial { "); + for (Types.NestedField field : table.schema().columns()) { + schema.append(parquetField(field)); + } + MessageType message = MessageTypeParser.parseMessageType(schema.append("}").toString()); + String[] payloads = {"0101000000000000000000F03F0000000000000040", + "00000000013FF00000000000004000000000000000", + "01b90b0000000000000000f03f000000000000004000000000000008400000000000001040", + "010200000000000000"}; + for (int id = 1; id <= 6; id++) { + Group row = new SimpleGroupFactory(message).newGroup().append("id", id); + for (String prefix : new String[] {"g_", "gg_"}) { + if (id == 5) { + continue; // SQL NULL parents. + } + Group array = row.addGroup(prefix + "array"); + Group map = row.addGroup(prefix + "map"); + Group structure = row.addGroup(prefix + "struct").append("label", "row" + id); + Group deep = row.addGroup(prefix + "deep"); + if (id == 6) { + continue; // Empty containers and a present struct with a NULL spatial field. + } + Binary point = binary(payloads[id - 1]); + array.addGroup("list").append("element", point); + array.addGroup("list"); + map.addGroup("key_value").append("key", "shape").append("value", point); + map.addGroup("key_value").append("key", "absent"); + structure.append("shape", point); + Group list = deep.addGroup("key_value").append("key", "shapes").addGroup("value"); + list.addGroup("list").addGroup("element").append("shape", point); + list.addGroup("list").addGroup("element"); + list.addGroup("list"); + } + appendParquetRow(table, message, row); + } + } + + private static void createOrcWriteTargets(Catalog catalog, Namespace namespace, + Map<String, String> properties) { + Map<String, String> orcProperties = new HashMap<>(properties); + orcProperties.put("write.format.default", "orc"); + orcProperties.put("write-format", "orc"); + for (boolean geography : new boolean[] {false, true}) { + for (boolean nested : new boolean[] {false, true}) { + TableIdentifier name = TableIdentifier.of(namespace, "spatial_existing_orc_" + + (geography ? "geog" : "geom") + (nested ? "_nested" : "")); + if (catalog.tableExists(name)) { + catalog.dropTable(name); + } + org.apache.iceberg.types.Type type = geography + ? Types.GeographyType.crs84() : Types.GeometryType.crs84(); + catalog.createTable(name, new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "shape", nested + ? Types.ListType.ofOptional(3, type) : type)), + PartitionSpec.unpartitioned(), orcProperties); + } + } + } + + private static void createEmbeddedNullCrsSources(Catalog catalog, Namespace namespace, + Map<String, String> properties) throws Exception { + for (boolean geography : new boolean[] {false, true}) { + String suffix = geography ? "geog" : "geom"; + // Geography's ')' survives the SDK parser. A C-string truncation would turn + // its serialized prefix into the valid, different geography(custom) type. + String crs = (geography ? "custom)" : "custom:") + (char) 0 + "zone"; + org.apache.iceberg.types.Type type = geography + ? Types.GeographyType.of(crs, org.apache.iceberg.types.EdgeAlgorithm.SPHERICAL) + : Types.GeometryType.of(crs); + Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "shape", type)); + for (boolean copy : new boolean[] {false, true}) { + TableIdentifier name = TableIdentifier.of(namespace, + "spatial_nul_crs_" + suffix + (copy ? "_copy" : "")); + if (catalog.tableExists(name)) { + catalog.dropTable(name); + } + Table table = catalog.createTable(name, schema, PartitionSpec.unpartitioned(), properties); + if (!copy) { + MessageType message = MessageTypeParser.parseMessageType( + "message spatial_nul_crs { required int32 id = 1; optional binary shape = 2; }"); + appendParquetRow(table, message, new SimpleGroupFactory(message).newGroup().append("id", 1) + .append("shape", binary("0101000000000000000000F03F0000000000000040"))); + } + } + } + } + + public static void main(String[] args) throws Exception { + Map<String, String> properties = new HashMap<>(); + properties.put("type", "rest"); + properties.put("uri", "http://rest:8181";); + properties.put("warehouse", "s3://warehouse/wh/"); + properties.put("io-impl", "org.apache.iceberg.aws.s3.S3FileIO"); + properties.put("s3.endpoint", "http://minio:9000";); + properties.put("s3.path-style-access", "true"); + properties.put("s3.region", "us-east-1"); + Catalog catalog = CatalogUtil.buildIcebergCatalog("demo", properties, null); + SupportsNamespaces namespaceCatalog = (SupportsNamespaces) catalog; + Namespace namespace = Namespace.of(args[0]); + if (!namespaceCatalog.namespaceExists(namespace)) { + namespaceCatalog.createNamespace(namespace); + } + TableIdentifier identifier = TableIdentifier.of(namespace, args[1]); + if (catalog.tableExists(identifier)) { + catalog.dropTable(identifier); + } + Schema schema = new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "geom", Types.GeometryType.crs84()), + Types.NestedField.optional(3, "geog", Types.GeographyType.crs84())); + Map<String, String> tableProperties = new HashMap<>(); + tableProperties.put("format-version", "3"); + tableProperties.put("write.format.default", "parquet"); + catalog.createTable(identifier, schema, PartitionSpec.unpartitioned(), tableProperties); + TableIdentifier nested = TableIdentifier.of(namespace, "nested_spatial_values"); + if (catalog.tableExists(nested)) { + catalog.dropTable(nested); + } + Table nestedTable = catalog.createTable(nested, new Schema( + Types.NestedField.optional(1, "shapes", + Types.ListType.ofOptional(2, Types.GeometryType.crs84()))), + PartitionSpec.unpartitioned(), tableProperties); + nestedTable.updateProperties().set("write.merge.mode", "merge-on-read").commit(); + TableIdentifier projected = TableIdentifier.of(namespace, "projected_values"); + if (catalog.tableExists(projected)) { + catalog.dropTable(projected); + } + Table projectedTable = catalog.createTable(projected, new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "geom", Types.GeometryType.of("EPSG:3857")), + Types.NestedField.optional(3, "geog", Types.GeographyType.of("OGC:CRS84", + org.apache.iceberg.types.EdgeAlgorithm.VINCENTY))), + PartitionSpec.unpartitioned(), tableProperties); + // A table property resolves the Iceberg CRS reference; it is unavailable to the BE file writer. + Map<String, String> referencedProperties = new HashMap<>(tableProperties); + referencedProperties.put("my_crs", "{\\"type\\":\\"GeographicCRS\\",\\"name\\":\\"WGS 84\\",\\"datum\\":{\\"type\\":\\"GeodeticReferenceFrame\\",\\"name\\":\\"World Geodetic System 1984\\",\\"ellipsoid\\":{\\"name\\":\\"WGS 84\\",\\"semi_major_axis\\":6378137,\\"inverse_flattening\\":298.257223563}},\\"coordinate_system\\":{\\"subtype\\":\\"ellipsoidal\\",\\"axis\\":[{\\"name\\":\\"Geodetic longitude\\",\\"abbreviation\\":\\"Lon\\",\\"direction\\":\\"east\\",\\"unit\\":\\"degree\\"},{\\"name\\":\\"Geodetic latitude\\",\\"abbreviation\\":\\"Lat\\",\\"direction\\":\\"north\\",\\"unit\\":\\"degree\\"}]}}"); + TableIdentifier referenced = TableIdentifier.of(namespace, "referenced_crs_values"); + if (catalog.tableExists(referenced)) { + catalog.dropTable(referenced); + } + Table referencedTable = catalog.createTable(referenced, new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "geom", Types.GeometryType.of("projjson:my_crs")), + Types.NestedField.optional(3, "geog", Types.GeographyType.of("projjson:my_crs", + org.apache.iceberg.types.EdgeAlgorithm.SPHERICAL))), + PartitionSpec.unpartitioned(), referencedProperties); + for (boolean isGeography : new boolean[] {false, true}) { + String column = isGeography ? "geog" : "geom"; + Schema copySchema = new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, column, isGeography + ? Types.GeographyType.of("projjson:my_crs", + org.apache.iceberg.types.EdgeAlgorithm.SPHERICAL) + : Types.GeometryType.of("projjson:my_crs"))); + for (boolean partitioned : new boolean[] {false, true}) { + String name = (isGeography ? "referenced_geography_copy" : "referenced_geometry_copy") + + (partitioned ? "_partitioned" : ""); + TableIdentifier copy = TableIdentifier.of(namespace, name); + if (catalog.tableExists(copy)) { + catalog.dropTable(copy); + } + Table copyTable = catalog.createTable(copy, copySchema, partitioned + ? PartitionSpec.builderFor(copySchema).identity("id").build() + : PartitionSpec.unpartitioned(), referencedProperties); + System.out.println("SPATIAL_DATA_LOCATION\\t" + name + "\\t" + + copyTable.properties().getOrDefault("write.data.path", copyTable.location() + "/data")); + } + } + // Iceberg 1.10 exposes spatial schema types but its generic Parquet writer does not implement them. + // Store standard WKB BYTE_ARRAY values with field ids; the Iceberg schema carries CRS/algorithm. + String parquetSchema = "message spatial { required int32 id = " + + projectedTable.schema().findField("id").fieldId() + "; optional binary geom = " + + projectedTable.schema().findField("geom").fieldId() + "; optional binary geog = " + + projectedTable.schema().findField("geog").fieldId() + "; }"; + MessageType message = MessageTypeParser.parseMessageType(parquetSchema); + Group row = new SimpleGroupFactory(message).newGroup().append("id", 1); + ByteBuffer point = ByteBuffer.allocate(21).order(ByteOrder.LITTLE_ENDIAN); + point.put((byte) 1).putInt(1).putDouble(1000).putDouble(2000).flip(); + row.append("geom", Binary.fromConstantByteBuffer(point)); + ByteBuffer geography = ByteBuffer.allocate(21).order(ByteOrder.LITTLE_ENDIAN); + geography.put((byte) 1).putInt(1).putDouble(1).putDouble(2).flip(); + row.append("geog", Binary.fromConstantByteBuffer(geography)); + appendParquetRow(projectedTable, message, row); + Group referencedRow = new SimpleGroupFactory(message).newGroup().append("id", 1); + referencedRow.append("geom", row.getBinary("geog", 0)); + referencedRow.append("geog", row.getBinary("geog", 0)); + appendParquetRow(referencedTable, message, referencedRow); + MessageType nestedMessage = MessageTypeParser.parseMessageType( + "message spatial_nested { optional group shapes (LIST) = 1 { " + + "repeated group list { optional binary element = 2; } } }"); + Group nestedRow = new SimpleGroupFactory(nestedMessage).newGroup(); + nestedRow.addGroup("shapes").addGroup("list").append("element", row.getBinary("geog", 0)); + appendParquetRow(nestedTable, nestedMessage, nestedRow); + createNestedSource(catalog, namespace, tableProperties); + createRequiredNestedTargets(catalog, namespace, tableProperties); + createOrcWriteTargets(catalog, namespace, tableProperties); + createEmbeddedNullCrsSources(catalog, namespace, tableProperties); + } +} +''' + String encodedJavaSource = javaSource.getBytes("UTF-8").encodeBase64().toString() + String fixtureOutput = executeCommand("${dockerCommand} exec ${sparkContainer} bash -lc '" + + "task_java_dir=\$(mktemp -d /tmp/iceberg-spatial-v3.XXXXXX) && " + + "echo ${encodedJavaSource} | base64 -d >\"\$task_java_dir/CreateIcebergSpatialV3Table.java\" && " + + "javac -cp \"/opt/spark/jars/*\" \"\$task_java_dir/CreateIcebergSpatialV3Table.java\" && " + + "java -cp \"\$task_java_dir:/opt/spark/jars/*\" " + + "CreateIcebergSpatialV3Table ${dbName} geometry_values'") + + def dataLocations = [:] + fixtureOutput.readLines().findAll { it.startsWith("SPATIAL_DATA_LOCATION\t") }.each { line -> + def fields = line.split("\t", 3) + dataLocations[fields[1]] = fields[2] + } + def minioClient = AmazonS3ClientBuilder.standard() + .withEndpointConfiguration(new EndpointConfiguration("http://${externalEnvIp}:${minioPort}";, "us-east-1")) + .withPathStyleAccessEnabled(true) + .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("admin", "password"))) + .build() + def listDataObjects = { String table -> + String location = dataLocations[table] + assertNotNull(location, "Missing data location for ${table}") + def uri = new URI(location) + def objects = new TreeMap<String, Long>() + String token = null + while (true) { + def listing = minioClient.listObjectsV2(new ListObjectsV2Request() + .withBucketName(uri.host).withPrefix(uri.path.substring(1).replaceAll('/+$', '') + '/') + .withContinuationToken(token)) + listing.objectSummaries.each { objects[it.key] = it.size } + if (!listing.isTruncated()) { + break + } + token = listing.nextContinuationToken + } + return objects + } + + def createCatalog = { + sql """drop catalog if exists ${catalogName}""" + sql """ + create catalog ${catalogName} properties ( + 'type' = 'iceberg', + 'iceberg.catalog.type' = 'rest', + 'uri' = 'http://${externalEnvIp}:${restPort}', + 's3.access_key' = 'admin', + 's3.secret_key' = 'password', + 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}', + 's3.region' = 'us-east-1' + ) + """ + sql """switch ${catalogName}""" + sql """use ${dbName}""" + } + createCatalog() + test { + sql """create table spatial_invalid_geometry_crs (geom geometry('')) + engine=iceberg properties ('format-version' = '3')""" + exception "Spatial CRS must not be empty" + } + test { + sql """create table spatial_invalid_geography_crs (geog geography('', 'spherical')) + engine=iceberg properties ('format-version' = '3')""" + exception "Spatial CRS must not be empty" + } + test { + sql """create table spatial_invalid_algorithm (geog geography('OGC:CRS84', 'unknown')) + engine=iceberg properties ('format-version' = '3')""" + exception "Unsupported Geography algorithm" + } + sql """ + insert into geometry_values values + (1, ST_GeometryFromWKBTyped('0101000000000000000000F03F0000000000000040'), + ST_GeogFromWKB('0101000000000000000000F03F0000000000000040')), + (2, NULL, NULL), + (3, ST_GeometryFromWKBTyped('010100000000000000000008400000000000001040'), NULL), + (4, NULL, ST_GeogFromWKB('010100000000000000000008400000000000001040')) + """ + // Doris writes OPTIONAL GEOMETRY/GEOGRAPHY annotations; count each leaf separately + // to exercise definition-level COUNT pushdown, including SQL NULL rows. + for (boolean pushdown : [false, true]) { + "qt_optional_geometry_count_${pushdown}"("""select /*+ SET_VAR( + enable_push_down_no_group_agg=${pushdown}) */ + count(geom) from geometry_values""") + "qt_optional_geography_count_${pushdown}"("""select /*+ SET_VAR( + enable_push_down_no_group_agg=${pushdown}) */ + count(geog) from geometry_values""") + } + qt_spatial_schema """desc geometry_values""" + // These factories do not implement scalar spatial values. Fail in FE, including + // analytic DISTINCT, which is rewritten to multi_distinct_count after binding. + for (String column : ["geom", "geog"]) { + for (String query : [ + "select ${column}, count(*) from geometry_values group by ${column}", + "select id, ${column}, count(*) from geometry_values group by id, ${column}", + "select ${column}, count(*) from geometry_values group by grouping sets ((${column}), ())", + "select distinct ${column} from geometry_values", + "select distinct id, ${column} from geometry_values"]) { + test { + sql query + exception "GROUP BY and DISTINCT do not support spatial keys" + } + } + for (String expression : ["min(${column})", "max(${column})", + "min(${column}) over ()", "max(${column}) over ()", + "count(distinct ${column})", "count(distinct id, ${column})"]) { + test { + sql "select ${expression} from geometry_values" + exception "does not support spatial arguments" + } + } + // Compare only implemented types; MAP_CONTAINS_KEY must not inspect spatial values. + for (String expression : [ + "map_contains_value(map(id, ${column}), ${column})", + "map_contains_value(map(id, ${column}), NULL)", + "map_contains_value(map(id, id), ${column})", + "map_contains_value(map(id, array(${column})), array(${column}))", + "map_contains_key(map(${column}, id), ${column})", + "map_contains_key(map(${column}, id), NULL)", + "map_contains_key(map(id, ${column}), ${column})", + "map_contains_entry(map(id, ${column}), id, ${column})", + "map_contains_entry(map(id, ${column}), id, NULL)", + "map_contains_entry(map(id, id), id, ${column})", + "map_contains_entry(map(id, id), ${column}, id)", + "map_contains_entry(map(${column}, id), NULL, id)", + "map_contains_entry(map(id, array(${column})), id, array(${column}))"]) { + test { + sql """select ${expression} from geometry_values""" + exception "does not support spatial comparison" Review Comment: Fixed in b58b2e21698. The lookup and MAP aggregate/combinator cases that construct a MAP with a spatial key now expect the inner `map does not support spatial keys` error. Cases with a valid MAP and an invalid lookup still expect the comparison error. The Groovy suite compiles and all 46 covered expressions match their expected messages in the FE analyzer. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/connector/converter/ConnectorSpatialWriteNormalizer.java: ########## @@ -0,0 +1,747 @@ +// 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.connector.converter; + +import org.apache.doris.catalog.Column; +import org.apache.doris.nereids.CTEContext; +import org.apache.doris.nereids.analyzer.UnboundFunction; +import org.apache.doris.nereids.properties.OrderKey; +import org.apache.doris.nereids.rules.analysis.ExpressionAnalyzer; +import org.apache.doris.nereids.rules.rewrite.PushProjectThroughUnion; +import org.apache.doris.nereids.trees.expressions.Alias; +import org.apache.doris.nereids.trees.expressions.CTEId; +import org.apache.doris.nereids.trees.expressions.Cast; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator; +import org.apache.doris.nereids.trees.expressions.functions.scalar.Array; +import org.apache.doris.nereids.trees.expressions.functions.scalar.Cardinality; +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.CreateStruct; +import org.apache.doris.nereids.trees.expressions.literal.ArrayLiteral; +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.StringLikeLiteral; +import org.apache.doris.nereids.trees.expressions.literal.StructLiteral; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.algebra.SetOperation.Qualifier; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTE; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEAnchor; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer; +import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.nereids.trees.plans.logical.LogicalLimit; +import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation; +import org.apache.doris.nereids.trees.plans.logical.LogicalPlan; +import org.apache.doris.nereids.trees.plans.logical.LogicalProject; +import org.apache.doris.nereids.trees.plans.logical.LogicalSort; +import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias; +import org.apache.doris.nereids.trees.plans.logical.LogicalUnion; +import org.apache.doris.nereids.trees.plans.logical.ProjectProcessor; +import org.apache.doris.nereids.types.ArrayType; +import org.apache.doris.nereids.types.DataType; +import org.apache.doris.nereids.types.GeographyType; +import org.apache.doris.nereids.types.GeometryType; +import org.apache.doris.nereids.types.MapType; +import org.apache.doris.nereids.types.StructType; +import org.apache.doris.nereids.util.ExpressionUtils; +import org.apache.doris.nereids.util.TypeCoercionUtils; + +import com.google.common.collect.ImmutableList; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.IntStream; + +/** Restores proven NULL producer values before spatial sink coercion. */ +public final class ConnectorSpatialWriteNormalizer { + private ConnectorSpatialWriteNormalizer() { + } + + public static LogicalPlan normalize(List<Column> targets, LogicalPlan source, CTEContext cteContext) { + return normalize(targets, source, IntStream.range(0, targets.size()).boxed() + .collect(ImmutableList.toImmutableList()), cteContext); + } + + /** Aligns data targets with source positions when a row-level sink also carries routing columns. */ + public static LogicalPlan normalize(List<Column> targets, LogicalPlan source, + List<Integer> sourceOrdinals, CTEContext cteContext) { + if (targets.stream().noneMatch(column -> column.getType().typeContainsSpatialType())) { + return source; + } + if (source instanceof LogicalUnion && ((LogicalUnion) source).getQualifier() == Qualifier.ALL) { + return normalizeUnionAll(targets, (LogicalUnion) source, sourceOrdinals, cteContext); + } + if (source instanceof LogicalSubQueryAlias) { + LogicalPlan child = (LogicalPlan) source.child(0); + LogicalPlan normalized = normalize(targets, child, sourceOrdinals, cteContext); + return normalized == child ? source : (LogicalPlan) source.withChildren(ImmutableList.of(normalized)); + } + if (source instanceof LogicalProject && isForwardingProject((LogicalProject<?>) source)) { + Plan input = source.child(0); + while (input instanceof LogicalSubQueryAlias) { + input = input.child(0); + } + Plan boundary = input; + while (boundary instanceof LogicalSubQueryAlias + || (boundary instanceof LogicalProject && isForwardingProject((LogicalProject<?>) boundary))) { + boundary = boundary.child(0); + } + if (boundary instanceof LogicalUnion && ((LogicalUnion) boundary).getQualifier() == Qualifier.ALL) { + // Reuse the existing safe forwarding/cast projection rule, including its ExprId handling. + Optional<Plan> pushed = ProjectProcessor.tryProcessProject( + ((LogicalProject<?>) source).getProjects(), input); + // Forwarding aliases can also repeat or prune outputs without moving calculations. + if (!pushed.isPresent() && input instanceof LogicalUnion + && ((LogicalProject<?>) source).getProjects().stream() + .allMatch(project -> project instanceof Slot + || (project instanceof Alias && project.child(0) instanceof Slot))) { + pushed = Optional.of(PushProjectThroughUnion.doPushProject( + ((LogicalProject<?>) source).getProjects(), (LogicalUnion) input)); + } + if (pushed.isPresent()) { + return normalize(targets, (LogicalPlan) pushed.get(), sourceOrdinals, cteContext); + } + } + } + Resolver resolver = new Resolver(source, cteContext); + List<NamedExpression> outputs = new ArrayList<>(source.getOutput()); + List<LogicalPlan> wrappers = new ArrayList<>(); + Set<ExprId> sortInputs = new HashSet<>(); + LogicalPlan producerInput = source; + while (producerInput instanceof LogicalSort || producerInput instanceof LogicalLimit) { + wrappers.add(producerInput); + for (Expression expression : producerInput.getExpressions()) { + // A direct CARDINALITY(map_slot) depends on the entry count, which retagging preserves. + if (!isMapCardinality(expression)) { + for (Slot input : expression.getInputSlots()) { + sortInputs.add(input.getExprId()); + } + } + } + producerInput = (LogicalPlan) producerInput.child(0); + } + LogicalProject<?> producer = producerInput instanceof LogicalProject + && !((LogicalProject<?>) producerInput).isDistinct() ? (LogicalProject<?>) producerInput : null; + List<NamedExpression> producerOutputs = null; + Map<ExprId, Slot> replacedSlots = new HashMap<>(); + boolean changed = false; + for (int i = 0; i < targets.size(); i++) { + Column target = targets.get(i); + if (!target.getType().typeContainsSpatialType()) { + continue; + } + int sourceOrdinal = sourceOrdinals.get(i); + Slot slot = source.getOutput().get(sourceOrdinal); + Expression value = resolver.resolve(slot, source); + DataType targetType = DataType.fromCatalogType(target.getType()); + boolean literal = isLiteralConstruction(value) || canRetargetSpatialNulls(value, targetType); + // SELECT/UPDATE reach the bound sink with CAST(ARRAY(NULL) AS ARRAY<INT>). + // Reuse MERGE's NULL-only proof before validation, without lifting arbitrary producers. + if (literal) { + value = normalizeSpatialNullsBeforeBinding(value, targetType); + } + if (!literal && producer != null && !sortInputs.contains(slot.getExprId()) + && sourceOrdinals.indexOf(sourceOrdinal) == sourceOrdinals.lastIndexOf(sourceOrdinal)) { + // Retag a constructor where it is computed, so its dynamic ordinary siblings + // still use the same inputs and are evaluated only once. Do not follow slots + // through joins, other producers, or a DISTINCT projection. + NamedExpression project = producer.getProjects().get(sourceOrdinal); + if (project instanceof Alias && canRetargetSpatialNulls(project.child(0), targetType, true)) { + Expression normalized = normalizeSpatialNullsBeforeBinding(project.child(0), targetType, true); + if (!normalized.getDataType().equals(slot.getDataType())) { + value = normalized; + if (producerOutputs == null) { + producerOutputs = new ArrayList<>(producer.getProjects()); + } + Alias replacement = new Alias(normalized, slot.getName()); + producerOutputs.set(sourceOrdinal, replacement); + Slot replacementSlot = replacement.toSlot(); + outputs.set(sourceOrdinal, replacementSlot); + replacedSlots.put(slot.getExprId(), replacementSlot); + } + } + } + ConnectorSpatialWriteValidator.validate(target, value); + // Never lift arbitrary producer expressions above their input scope, or + // replace outer-join NULL padding with a non-NULL container constant. + if (literal) { + Expression normalized = normalizeSpatialNulls(value, targetType); + if (!normalized.getDataType().equals(slot.getDataType())) { + // SELECT v,v can feed an ordinary BIGINT and a spatial NULL. Sharing + // v's ExprId would let project merging replace the ordinary cast's input. + outputs.set(sourceOrdinal, new Alias(normalized, slot.getName())); + changed = true; + } + } + } + LogicalPlan normalizedSource = source; + if (producerOutputs != null) { + normalizedSource = producer.withProjects(producerOutputs); + for (int i = wrappers.size() - 1; i >= 0; i--) { + LogicalPlan wrapper = wrappers.get(i); + if (wrapper instanceof LogicalSort) { + LogicalSort<?> sort = (LogicalSort<?>) wrapper; + List<OrderKey> keys = sort.getOrderKeys().stream().map(key -> { + Expression expr = key.getExpr(); + if (isMapCardinality(expr)) { + Slot input = replacedSlots.get(((Slot) expr.child(0)).getExprId()); + if (input != null) { + return key.withExpression(new Cardinality(input)); + } + } + return key; + }).collect(ImmutableList.toImmutableList()); + normalizedSource = sort.withOrderKeysAndChild(keys, normalizedSource); + } else { + normalizedSource = (LogicalPlan) wrapper.withChildren(ImmutableList.of(normalizedSource)); + } + } + } + return changed ? new LogicalProject<>(outputs, normalizedSource) : normalizedSource; + } + + private static boolean isMapCardinality(Expression expression) { + return expression instanceof Cardinality && expression.child(0) instanceof Slot + && expression.child(0).getDataType() instanceof MapType; + } + + private static boolean isForwardingProject(LogicalProject<?> project) { + return !project.isDistinct() && project.getProjects().stream().allMatch(output -> output instanceof Slot + || (output instanceof Alias && ExpressionUtils.getExpressionCoveredByCast(output.child(0)) + instanceof Slot)); + } + + private static LogicalPlan normalizeUnionAll(List<Column> targets, LogicalUnion source, + List<Integer> sourceOrdinals, CTEContext context) { + List<Plan> children = new ArrayList<>(); + List<List<SlotReference>> childOutputs = new ArrayList<>(); + boolean changed = !source.getConstantExprsList().isEmpty(); + for (int i = 0; i < source.arity(); i++) { + LogicalPlan child = (LogicalPlan) source.child(i); + List<SlotReference> outputs = source.getRegularChildOutput(i); + LogicalPlan input = child.getOutput().equals(outputs) + ? child : new LogicalProject<>(ImmutableList.<NamedExpression>copyOf(outputs), child); + LogicalPlan normalized = normalize(targets, input, sourceOrdinals, context); + children.add(normalized); + childOutputs.add(normalized.getOutput().stream().map(SlotReference.class::cast) + .collect(ImmutableList.toImmutableList())); + changed |= normalized != child; + } + // Constant rows use the same proof and CAST validation as ordinary branches. + // Keep them as separate rows; never replace an entire UNION output with one constant. + for (List<NamedExpression> row : source.getConstantExprsList()) { + LogicalPlan normalized = normalize(targets, + new LogicalOneRowRelation(StatementScopeIdGenerator.newRelationId(), row), + sourceOrdinals, context); + children.add(normalized); + childOutputs.add(normalized.getOutput().stream().map(SlotReference.class::cast) + .collect(ImmutableList.toImmutableList())); + } + if (!changed) { + return source; + } + List<NamedExpression> outputs = new ArrayList<>(); + for (int i = 0; i < source.getOutput().size(); i++) { + SlotReference slot = (SlotReference) source.getOutput().get(i); + DataType type = childOutputs.get(0).get(i).getDataType(); Review Comment: Fixed in b58b2e21698. After spatial NULL retagging, UNION ALL computes the set-operation common type for each output and applies the required casts to every branch, including constant rows. The regression covers both regular-branch orders and a constant branch. It reproduces the type mismatch on the previous implementation and passes with the fix. ########## be/src/format_v2/table/iceberg_reader.h: ########## @@ -74,6 +74,7 @@ class IcebergTableReader : public format::TableReader { protected: // Iceberg UUID uses the same 16-byte STRING/VARBINARY carrier in data, defaults and deletes. bool preserve_binary_uuid() const override { return true; } + bool parquet_spatial_annotations_as_bytes() const override { return false; } Review Comment: Fixed in b58b2e21698. Iceberg uses a raw-byte Parquet read type for spatially annotated leaves mapped to table BINARY fields. Fields declared GEOMETRY or GEOGRAPHY keep typed decoding. The actual-file test covers scalar and nested fields matched by ID, both annotations, opaque TRIANGLE WKB and NULLs. The Iceberg and Parquet BE test suites pass. -- 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]
