Skip to content

Commit ef8fcba

Browse files
cwdobbinscwdobbins
authored
GEOMESA-3584 Trino - Read geometry type from table metadata (#3590)
1 parent 7f32595 commit ef8fcba

6 files changed

Lines changed: 285 additions & 32 deletions

File tree

‎geomesa-trino/geomesa-trino-datastore/pom.xml‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -62,9 +62,11 @@
6262
<dependency>
6363
<groupId>org.locationtech.geomesa</groupId>
6464
<artifactId>geomesa-security_${scala.binary.version}</artifactId>
65-
<version>${project.version}</version>
6665
</dependency>
67-
66+
<dependency>
67+
<groupId>org.locationtech.geomesa</groupId>
68+
<artifactId>geomesa-utils_${scala.binary.version}</artifactId>
69+
</dependency>
6870
<dependency>
6971
<groupId>org.junit.jupiter</groupId>
7072
<artifactId>junit-jupiter</artifactId>

‎geomesa-trino/geomesa-trino-datastore/src/main/java/org/locationtech/geomesa/trino/datastore/TrinoSchemaDiscovery.java‎

Lines changed: 92 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -9,14 +9,31 @@
99
package org.locationtech.geomesa.trino.datastore;
1010

1111
import org.geotools.api.feature.simple.SimpleFeatureType;
12+
import org.geotools.api.feature.type.AttributeDescriptor;
13+
import org.geotools.api.feature.type.GeometryDescriptor;
1214
import org.geotools.feature.simple.SimpleFeatureTypeBuilder;
15+
import org.locationtech.geomesa.utils.geotools.SimpleFeatureTypes;
16+
import org.locationtech.jts.geom.Geometry;
17+
import org.locationtech.jts.geom.Point;
18+
import org.slf4j.Logger;
19+
import org.slf4j.LoggerFactory;
1320

1421
import java.io.IOException;
1522
import java.sql.*;
1623
import java.util.*;
1724

25+
import static org.locationtech.geomesa.trino.datastore.TrinoDataStore.escapeQuotes;
26+
1827
class TrinoSchemaDiscovery {
1928

29+
private static final Logger LOG = LoggerFactory.getLogger(TrinoSchemaDiscovery.class);
30+
31+
/** Iceberg table property holding the GeoMesa-encoded SimpleFeatureType spec. */
32+
static final String SFT_SPEC_PROPERTY = "geomesa.sft.spec";
33+
34+
/** Iceberg table property holding the GeoMesa type name. */
35+
static final String SFT_NAME_PROPERTY = "geomesa.sft.name";
36+
2037
private final TrinoDataStore store;
2138

2239
TrinoSchemaDiscovery(TrinoDataStore store) {
@@ -28,8 +45,17 @@ SimpleFeatureType discover(String typeName) throws IOException {
2845
tb.setName(typeName);
2946
tb.setNamespaceURI(store.getNamespaceURI());
3047

31-
String sql = String.format("SELECT * FROM \"%s\".\"%s\".\"%s\" LIMIT 0",
32-
store.catalog(), store.trinoSchema(), typeName);
48+
String sql = String.format(
49+
"SELECT * FROM %s.%s.%s LIMIT 0",
50+
escapeQuotes(store.catalog()),
51+
escapeQuotes(store.trinoSchema()),
52+
escapeQuotes(typeName)
53+
);
54+
55+
Map<String, String> sftProps = readSftProperties(typeName);
56+
String sftSpec = sftProps.get(SFT_SPEC_PROPERTY);
57+
Map<String, Class<?>> sftBindings = (sftSpec == null || sftSpec.isBlank())
58+
? Map.of() : geometryBindingsFromSpec(typeName, sftSpec);
3359

3460
String visColumn = null;
3561
try (Connection conn = store.connect();
@@ -54,16 +80,14 @@ SimpleFeatureType discover(String typeName) throws IOException {
5480
for (int i = 1; i <= meta.getColumnCount(); i++) {
5581
String name = meta.getColumnName(i);
5682
if (TrinoTypeMapper.isHidden(name)) continue;
57-
if (name.equals(visColumn)) continue; // vis column is metadata, not a SFT attribute
83+
if (name.equals(visColumn)) continue; // vis column is metadata, not an SFT attribute
5884

5985
boolean isGeom = geometryColumnNames.contains(name);
60-
// A __X_z2__ companion marks a point-only geometry column, so the
61-
// attribute can be bound to Point rather than the generic Geometry.
62-
boolean isPoint = isGeom && isPointColumn(name, allNames);
86+
Class<?> geomBinding = isGeom ? resolveGeometryBinding(name, allNames, sftBindings) : null;
6387
// SRID is no longer carried in a table property; default to WGS84.
6488
int srid = isGeom ? 4326 : 0;
6589
var descriptor =
66-
TrinoTypeMapper.toDescriptor(name, meta.getColumnType(i), isGeom, isPoint, srid);
90+
TrinoTypeMapper.toDescriptor(name, meta.getColumnType(i), isGeom, geomBinding, srid);
6791
tb.add(descriptor);
6892

6993
if (isGeom && !defaultGeomSet) {
@@ -79,9 +103,70 @@ SimpleFeatureType discover(String typeName) throws IOException {
79103
if (visColumn != null) {
80104
sft.getUserData().put(VIS_COLUMN_KEY, visColumn);
81105
}
106+
String sftName = sftProps.get(SFT_NAME_PROPERTY);
107+
if (sftName != null && !sftName.isBlank()) {
108+
sft.getUserData().put(SFT_NAME_PROPERTY, sftName);
109+
}
82110
return sft;
83111
}
84112

113+
/** The JTS geometry binding for a geometry column: the subtype declared by the stored SFT
114+
* when it names this attribute, else {@link Point} for a {@code __<name>_z2__} companion,
115+
* generic {@link Geometry} otherwise. */
116+
static Class<?> resolveGeometryBinding(String name, Set<String> allNames,
117+
Map<String, Class<?>> sftBindings) {
118+
Class<?> fromSft = sftBindings.get(name);
119+
if (fromSft != null) {
120+
return fromSft;
121+
}
122+
return isPointColumn(name, allNames) ? Point.class : Geometry.class;
123+
}
124+
125+
/** Parses a GeoMesa SFT spec into geometry-attribute-name → JTS subtype. */
126+
static Map<String, Class<?>> geometryBindingsFromSpec(String typeName, String spec) {
127+
try {
128+
SimpleFeatureType sft = SimpleFeatureTypes.createType(typeName, spec);
129+
Map<String, Class<?>> bindings = new LinkedHashMap<>();
130+
for (AttributeDescriptor d : sft.getAttributeDescriptors()) {
131+
if (d instanceof GeometryDescriptor) {
132+
bindings.put(d.getLocalName(), d.getType().getBinding());
133+
}
134+
}
135+
return bindings;
136+
} catch (RuntimeException e) {
137+
LOG.warn("Could not parse stored GeoMesa SFT for '{}' ({}='{}'): {}; "
138+
+ "falling back to heuristic geometry binding",
139+
typeName, SFT_SPEC_PROPERTY, spec, e.getMessage());
140+
return Map.of();
141+
}
142+
}
143+
144+
/** Reads the {@code geomesa.sft.*} properties (spec + name) from the Iceberg
145+
* {@code <table>$properties} metadata table in one query; empty when the table carries none
146+
* or doesn't expose {@code $properties}. Non-fatal: any failure just disables SFT-driven
147+
* binding/name for this table. */
148+
private Map<String, String> readSftProperties(String typeName) {
149+
String sql = String.format(
150+
"SELECT key, value FROM %s.%s.%s WHERE key IN ('%s', '%s')",
151+
escapeQuotes(store.catalog()),
152+
escapeQuotes(store.trinoSchema()),
153+
escapeQuotes(typeName + "$properties"),
154+
SFT_SPEC_PROPERTY, SFT_NAME_PROPERTY
155+
);
156+
Map<String, String> props = new HashMap<>();
157+
try (Connection conn = store.connect();
158+
Statement stmt = conn.createStatement();
159+
ResultSet rs = stmt.executeQuery(sql)) {
160+
while (rs.next()) {
161+
props.put(rs.getString(1), rs.getString(2));
162+
}
163+
} catch (SQLException e) {
164+
LOG.debug("No readable $properties for '{}' (no SFT metadata): {}",
165+
typeName, e.getMessage());
166+
}
167+
return props;
168+
}
169+
85170
/** Returns the set of column names that are geometry columns under the
86171
* naming convention: a base name {@code X} is a geometry column iff at least
87172
* one of {@code __X_bbox__}, {@code __X_z2__}, {@code __X_xz2__} appears in

‎geomesa-trino/geomesa-trino-datastore/src/main/java/org/locationtech/geomesa/trino/datastore/TrinoTypeMapper.java‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,6 @@
1212
import org.geotools.feature.AttributeTypeBuilder;
1313
import org.geotools.referencing.crs.DefaultGeographicCRS;
1414
import org.locationtech.jts.geom.Geometry;
15-
import org.locationtech.jts.geom.Point;
1615

1716
import java.sql.Types;
1817
import java.util.Date;
@@ -36,7 +35,7 @@ static boolean isHidden(String columnName) {
3635
}
3736

3837
static AttributeDescriptor toDescriptor(String name, int sqlType,
39-
boolean isGeometry, boolean isPoint, int srid) {
38+
boolean isGeometry, Class<?> geometryBinding, int srid) {
4039
AttributeTypeBuilder b = new AttributeTypeBuilder();
4140
b.setName(name);
4241
b.setNillable(true);
@@ -52,10 +51,9 @@ static AttributeDescriptor toDescriptor(String name, int sqlType,
5251
}
5352
}
5453
b.setCRS(crs);
55-
// Point when the column carries a __<name>_z2__ companion; this enables the
56-
// point/rectangle bbox fast path in TrinoFilterToSQL. Generic Geometry
57-
// otherwise (XZ2-companioned columns may hold any geometry type).
58-
b.setBinding(isPoint ? Point.class : Geometry.class);
54+
Class<?> binding = geometryBinding != null && Geometry.class.isAssignableFrom(geometryBinding)
55+
? geometryBinding : Geometry.class;
56+
b.setBinding(binding);
5957
return b.buildDescriptor(name, b.buildGeometryType());
6058
}
6159

‎geomesa-trino/geomesa-trino-datastore/src/test/java/org/locationtech/geomesa/trino/datastore/GeometrySubtypeIT.java‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
/***********************************************************************
2+
* Copyright (c) 2013-2025 General Atomics Integrated Intelligence, Inc.
3+
* All rights reserved. This program and the accompanying materials
4+
* are made available under the terms of the Apache License, Version 2.0
5+
* which accompanies this distribution and is available at
6+
* https://www.apache.org/licenses/LICENSE-2.0
7+
***********************************************************************/
8+
9+
package org.locationtech.geomesa.trino.datastore;
10+
11+
import org.geotools.api.data.DataStore;
12+
import org.geotools.api.data.DataStoreFinder;
13+
import org.geotools.api.feature.simple.SimpleFeatureType;
14+
import org.junit.jupiter.api.AfterAll;
15+
import org.junit.jupiter.api.Assumptions;
16+
import org.junit.jupiter.api.BeforeAll;
17+
import org.junit.jupiter.api.Tag;
18+
import org.junit.jupiter.api.Test;
19+
import org.locationtech.jts.geom.Point;
20+
import org.locationtech.jts.geom.Polygon;
21+
22+
import java.util.Arrays;
23+
import java.util.Map;
24+
import java.util.Set;
25+
import java.util.stream.Collectors;
26+
27+
import static org.assertj.core.api.Assertions.assertThat;
28+
29+
/**
30+
* End-to-end check for GEOMESA-3584: when an Iceberg table carries a GeoMesa-encoded SFT
31+
* (written by the tooling ingest as the {@code geomesa.sft.spec} table property), schema
32+
* discovery binds each geometry attribute to the subtype the SFT declares.
33+
*
34+
* <p>Requires a running Trino at localhost:8080 with the synthetic demo re-ingested by a build
35+
* that writes the SFT property (the {@code spatial.regions} table is XZ2-partitioned, so the
36+
* naming heuristic alone would bind it to generic {@code Geometry} — binding {@code Polygon}
37+
* proves the SFT drove it). Skips when the tables are absent.
38+
*/
39+
@Tag("integration")
40+
class GeometrySubtypeIT {
41+
42+
private static DataStore ds;
43+
private static Set<String> tables = Set.of();
44+
45+
@BeforeAll
46+
static void connect() throws Exception {
47+
try (var socket = new java.net.Socket()) {
48+
socket.connect(new java.net.InetSocketAddress("localhost", 8080), 2000);
49+
} catch (Exception e) {
50+
Assumptions.assumeTrue(false, "Trino not reachable at localhost:8080 — skipping");
51+
}
52+
ds = DataStoreFinder.getDataStore(Map.of(
53+
"trino.host", "localhost", "trino.port", 8080,
54+
"trino.catalog", "spatial_iceberg", "trino.schema", "spatial"));
55+
assertThat(ds).as("DataStoreFinder must locate TrinoDataStoreFactory").isNotNull();
56+
tables = Arrays.stream(ds.getTypeNames()).collect(Collectors.toSet());
57+
}
58+
59+
@AfterAll
60+
static void disconnect() {
61+
if (ds != null) ds.dispose();
62+
}
63+
64+
@Test
65+
void xz2PolygonColumnBindsToPolygonFromSft() throws Exception {
66+
Assumptions.assumeTrue(tables.contains("regions"),
67+
"spatial.regions not ingested (needs an SFT-writing ingest) — skipping");
68+
SimpleFeatureType sft = ds.getSchema("regions");
69+
assertThat(sft.getGeometryDescriptor().getType().getBinding())
70+
.as("XZ2 column bound to its SFT-declared subtype, not generic Geometry")
71+
.isEqualTo(Polygon.class);
72+
}
73+
74+
@Test
75+
void sftNameRecordedInUserData() throws Exception {
76+
Assumptions.assumeTrue(tables.contains("regions"),
77+
"spatial.regions not ingested (needs an SFT-writing ingest) — skipping");
78+
SimpleFeatureType sft = ds.getSchema("regions");
79+
assertThat(sft.getUserData().get(TrinoSchemaDiscovery.SFT_NAME_PROPERTY))
80+
.as("original GeoMesa type name recorded from geomesa.sft.name")
81+
.isEqualTo("regions");
82+
}
83+
84+
@Test
85+
void z2PointColumnRemainsPoint() throws Exception {
86+
Assumptions.assumeTrue(tables.contains("observations"),
87+
"spatial.observations not ingested — skipping");
88+
SimpleFeatureType sft = ds.getSchema("observations");
89+
assertThat(sft.getGeometryDescriptor().getType().getBinding()).isEqualTo(Point.class);
90+
}
91+
92+
@Test
93+
void multiGeomBindsEachSubtypeIndependently() throws Exception {
94+
Assumptions.assumeTrue(tables.contains("observations_2geom"),
95+
"spatial.observations_2geom not ingested — skipping");
96+
SimpleFeatureType sft = ds.getSchema("observations_2geom");
97+
assertThat(sft.getDescriptor("center").getType().getBinding()).isEqualTo(Point.class);
98+
assertThat(sft.getDescriptor("ellipse").getType().getBinding()).isEqualTo(Polygon.class);
99+
}
100+
}

‎geomesa-trino/geomesa-trino-datastore/src/test/java/org/locationtech/geomesa/trino/datastore/TrinoSchemaDiscoveryTest.java‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,11 @@
99
package org.locationtech.geomesa.trino.datastore;
1010

1111
import org.junit.jupiter.api.Test;
12+
import org.locationtech.jts.geom.Geometry;
13+
import org.locationtech.jts.geom.Point;
14+
import org.locationtech.jts.geom.Polygon;
1215

16+
import java.util.Map;
1317
import java.util.Set;
1418

1519
import static org.assertj.core.api.Assertions.assertThat;
@@ -47,4 +51,50 @@ void xz2CompanionIsNotPointColumn() {
4751
assertThat(TrinoSchemaDiscovery.isPointColumn("ellipse",
4852
Set.of("ellipse", "__ellipse_xz2__", "__ellipse_bbox__"))).isFalse();
4953
}
54+
55+
@Test
56+
void sftBindingWinsOverHeuristic() {
57+
// The stored SFT declares the exact subtype; it overrides the companion heuristic —
58+
// both directions: an XZ2 column (heuristic would say generic Geometry) bound to
59+
// Polygon, and a Z2 column (heuristic would say Point) bound to whatever the SFT says.
60+
var allNames = Set.of("region", "__region_xz2__", "__region_bbox__");
61+
assertThat(TrinoSchemaDiscovery.resolveGeometryBinding(
62+
"region", allNames, Map.of("region", Polygon.class))).isEqualTo(Polygon.class);
63+
assertThat(TrinoSchemaDiscovery.resolveGeometryBinding(
64+
"center", Set.of("center", "__center_z2__"), Map.of("center", Polygon.class)))
65+
.isEqualTo(Polygon.class);
66+
}
67+
68+
@Test
69+
void parsesSftWithOptionsAndUserData() {
70+
String spec = "id:String:fs.bounds=true,dtg:Date:default=true:fs.bounds=true,"
71+
+ "name:String,count:Long,score:Double:fs.bounds=true,active:Boolean,"
72+
+ "*geom:Point:srid=4326;geomesa.index.dtg='dtg'";
73+
Map<String, Class<?>> bindings =
74+
TrinoSchemaDiscovery.geometryBindingsFromSpec("example", spec);
75+
assertThat(bindings).containsExactly(Map.entry("geom", Point.class));
76+
}
77+
78+
@Test
79+
void multiGeomSpecYieldsEachSubtype() {
80+
Map<String, Class<?>> bindings = TrinoSchemaDiscovery.geometryBindingsFromSpec(
81+
"t", "*center:Point:srid=4326,ellipse:Polygon:srid=4326,dtg:Date,label:String");
82+
assertThat(bindings).containsOnly(
83+
Map.entry("center", Point.class), Map.entry("ellipse", Polygon.class));
84+
}
85+
86+
@Test
87+
void unparseableSpecYieldsEmptyForHeuristicFallback() {
88+
assertThat(TrinoSchemaDiscovery.geometryBindingsFromSpec("t", "]not a valid spec["))
89+
.isEmpty();
90+
}
91+
92+
@Test
93+
void fallsBackToHeuristicWhenSftAbsent() {
94+
// No SFT entry for this column → z2 companion ⇒ Point, xz2 companion ⇒ generic Geometry.
95+
assertThat(TrinoSchemaDiscovery.resolveGeometryBinding(
96+
"center", Set.of("center", "__center_z2__"), Map.of())).isEqualTo(Point.class);
97+
assertThat(TrinoSchemaDiscovery.resolveGeometryBinding(
98+
"ellipse", Set.of("ellipse", "__ellipse_xz2__"), Map.of())).isEqualTo(Geometry.class);
99+
}
50100
}

0 commit comments

Comments
 (0)