diff --git a/docs/static/generated/rest_v1_sql_gateway.yml b/docs/static/generated/rest_v1_sql_gateway.yml index fecf0576a011cd..baee191c2a042d 100644 --- a/docs/static/generated/rest_v1_sql_gateway.yml +++ b/docs/static/generated/rest_v1_sql_gateway.yml @@ -403,6 +403,7 @@ components: - DESCRIPTOR - VARIANT - BITMAP + - GEOGRAPHY OpenSessionRequestBody: type: object properties: diff --git a/docs/static/generated/rest_v2_sql_gateway.yml b/docs/static/generated/rest_v2_sql_gateway.yml index e39feaaea8ed53..0b6b70219283b4 100644 --- a/docs/static/generated/rest_v2_sql_gateway.yml +++ b/docs/static/generated/rest_v2_sql_gateway.yml @@ -477,6 +477,7 @@ components: - DESCRIPTOR - VARIANT - BITMAP + - GEOGRAPHY OpenSessionRequestBody: type: object properties: diff --git a/docs/static/generated/rest_v3_sql_gateway.yml b/docs/static/generated/rest_v3_sql_gateway.yml index 6d171ac2152f37..bc46614b4248b6 100644 --- a/docs/static/generated/rest_v3_sql_gateway.yml +++ b/docs/static/generated/rest_v3_sql_gateway.yml @@ -506,6 +506,7 @@ components: - DESCRIPTOR - VARIANT - BITMAP + - GEOGRAPHY OpenSessionRequestBody: type: object properties: diff --git a/docs/static/generated/rest_v4_sql_gateway.yml b/docs/static/generated/rest_v4_sql_gateway.yml index d39da5e007c2b7..b1d917452af648 100644 --- a/docs/static/generated/rest_v4_sql_gateway.yml +++ b/docs/static/generated/rest_v4_sql_gateway.yml @@ -516,6 +516,7 @@ components: - DESCRIPTOR - VARIANT - BITMAP + - GEOGRAPHY OpenSessionRequestBody: type: object properties: diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncRunnableStreamOperatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncRunnableStreamOperatorTest.java index b073f6cbed1210..1601b06bda9986 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncRunnableStreamOperatorTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncRunnableStreamOperatorTest.java @@ -253,8 +253,11 @@ void testCheckpointDrain() throws Exception { }); ((AbstractAsyncRunnableStreamOperator) testHarness.getOperator()) .postProcessElement(); - assertThat(asyncExecutionController.getInFlightRecordNum()).isEqualTo(1); - unblockAsyncRequest.complete(null); + try { + assertThat(asyncExecutionController.getInFlightRecordNum()).isEqualTo(1); + } finally { + unblockAsyncRequest.complete(null); + } testHarness.drainAsyncRequests(); assertThat(asyncExecutionController.getInFlightRecordNum()).isEqualTo(0); } diff --git a/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd b/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd index b7e6d7ab9ea3f4..db08bbfbf2c329 100644 --- a/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd +++ b/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd @@ -173,6 +173,7 @@ "org.apache.flink.sql.parser.type.ExtendedSqlCollectionTypeNameSpec" "org.apache.flink.sql.parser.type.ExtendedSqlRowTypeNameSpec" "org.apache.flink.sql.parser.type.SqlBitmapTypeNameSpec" + "org.apache.flink.sql.parser.type.SqlGeographyTypeNameSpec" "org.apache.flink.sql.parser.type.SqlRawTypeNameSpec" "org.apache.flink.sql.parser.type.SqlStructuredTypeNameSpec" "org.apache.flink.sql.parser.type.SqlTimestampLtzTypeNameSpec" @@ -225,6 +226,7 @@ "FROM_TIMESTAMP" "FUNCTIONS" "FRESHNESS" + "GEOGRAPHY" "HASH" "IF" "JSON_EXECUTION_PLAN" @@ -719,6 +721,7 @@ "ExtendedSqlRowTypeName()" "SqlStructuredTypeName()" "SqlBitmapTypeName()" + "SqlGeographyTypeName()" ] # List of methods for parsing builtin function calls. diff --git a/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl b/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl index 2b8250ea98ed9e..ede54c27ca86e0 100644 --- a/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl +++ b/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl @@ -2666,6 +2666,17 @@ SqlTypeNameSpec SqlBitmapTypeName() : } } +/** Parses GEOGRAPHY type. */ +SqlTypeNameSpec SqlGeographyTypeName() : +{ +} +{ + + { + return new SqlGeographyTypeNameSpec(getPos()); + } +} + /** * Parse a "name1 type1 [ NULL | NOT NULL] [ comment ] * [, name2 type2 [ NULL | NOT NULL] [ comment ] ]* ..." list. diff --git a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/type/SqlGeographyTypeNameSpec.java b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/type/SqlGeographyTypeNameSpec.java new file mode 100644 index 00000000000000..55a3d7ff4fd7d6 --- /dev/null +++ b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/type/SqlGeographyTypeNameSpec.java @@ -0,0 +1,59 @@ +/* + * 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.flink.sql.parser.type; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.calcite.ExtendedRelTypeFactory; + +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.sql.SqlIdentifier; +import org.apache.calcite.sql.SqlTypeNameSpec; +import org.apache.calcite.sql.SqlWriter; +import org.apache.calcite.sql.parser.SqlParserPos; +import org.apache.calcite.sql.validate.SqlValidator; +import org.apache.calcite.util.Litmus; + +/** Represents the GEOGRAPHY data type. */ +@Internal +public final class SqlGeographyTypeNameSpec extends SqlTypeNameSpec { + + private static final String GEOGRAPHY_TYPE_NAME = "GEOGRAPHY"; + + public SqlGeographyTypeNameSpec(SqlParserPos pos) { + super(new SqlIdentifier(GEOGRAPHY_TYPE_NAME, pos), pos); + } + + @Override + public RelDataType deriveType(SqlValidator validator) { + return ((ExtendedRelTypeFactory) validator.getTypeFactory()).createGeographyType(); + } + + @Override + public void unparse(SqlWriter writer, int leftPrec, int rightPrec) { + writer.keyword(GEOGRAPHY_TYPE_NAME); + } + + @Override + public boolean equalsDeep(SqlTypeNameSpec spec, Litmus litmus) { + if (!(spec instanceof SqlGeographyTypeNameSpec)) { + return litmus.fail("{} != {}", this, spec); + } + return litmus.succeed(); + } +} diff --git a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/table/calcite/ExtendedRelTypeFactory.java b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/table/calcite/ExtendedRelTypeFactory.java index 57b9bc615ec930..57be54e4cb6bd4 100644 --- a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/table/calcite/ExtendedRelTypeFactory.java +++ b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/table/calcite/ExtendedRelTypeFactory.java @@ -44,4 +44,7 @@ RelDataType createStructuredType( /** Creates a BITMAP type. */ RelDataType createBitmapType(); + + /** Creates a GEOGRAPHY type. */ + RelDataType createGeographyType(); } diff --git a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java index 92f8e34c770a00..ee33ea2e37daf2 100644 --- a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java +++ b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java @@ -4074,4 +4074,23 @@ void testBitmapType() { sql("CREATE TABLE t (\n" + "^bitmap^ INT" + "\n)") .fails("(?s).*Encountered \"bitmap\" at line 2, column 1.\n.*"); } + + @Test + void testGeographyType() { + sql("CREATE TABLE t (\n" + "g geography" + "\n)") + .ok("CREATE TABLE `T` (\n" + " `G` GEOGRAPHY\n" + ")"); + + sql("CREATE TABLE t (\n" + "g geography NOT NULL" + "\n)") + .ok("CREATE TABLE `T` (\n" + " `G` GEOGRAPHY NOT NULL\n" + ")"); + + // GEOGRAPHY takes no parameters + sql("CREATE TABLE t (\n" + "g geography^(^1)" + "\n)") + .fails("(?s).*Encountered \"\\(\" at line 2, column 12.\n.*"); + sql("CREATE TABLE t (\n" + "g geography^(^4326)" + "\n)") + .fails("(?s).*Encountered \"\\(\" at line 2, column 12.\n.*"); + + // GEOGRAPHY is a reserved keyword and cannot be used as an identifier without escaping + sql("CREATE TABLE t (\n" + "^geography^ INT" + "\n)") + .fails("(?s).*Encountered \"geography\" at line 2, column 1.\n.*"); + } } diff --git a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/TestRelDataTypeFactory.java b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/TestRelDataTypeFactory.java index 34fe9926079b3c..070c77905a335f 100644 --- a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/TestRelDataTypeFactory.java +++ b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/TestRelDataTypeFactory.java @@ -53,6 +53,11 @@ public RelDataType createBitmapType() { return canonize(new DummyBitmapType()); } + @Override + public RelDataType createGeographyType() { + return canonize(new DummyGeographyType()); + } + private static class DummyRawType extends RelDataTypeImpl { private final String className; @@ -117,4 +122,16 @@ protected void generateTypeString(StringBuilder sb, boolean withDetail) { sb.append("BITMAP"); } } + + private static class DummyGeographyType extends RelDataTypeImpl { + + DummyGeographyType() { + computeDigest(); + } + + @Override + protected void generateTypeString(StringBuilder sb, boolean withDetail) { + sb.append("GEOGRAPHY"); + } + } } diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java index af1e718d90e04c..3912c178328526 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java @@ -44,6 +44,7 @@ import org.apache.flink.table.types.logical.DescriptorType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.LogicalType; @@ -1075,6 +1076,15 @@ public static DataType BITMAP() { return new AtomicDataType(new BitmapType()); } + /** + * Data type of geography data. + * + * @see GeographyType + */ + public static DataType GEOGRAPHY() { + return new AtomicDataType(new GeographyType()); + } + // -------------------------------------------------------------------------------------------- // Helper functions // -------------------------------------------------------------------------------------------- diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/ArrayData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/ArrayData.java index 811c9be70e6b87..728d3c228f05da 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/ArrayData.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/ArrayData.java @@ -123,6 +123,12 @@ default Bitmap getBitmap(int pos) { "This ArrayData implementation does not support Bitmap type."); } + /** Returns the geography value at the given position. */ + default GeographyData getGeography(int pos) { + throw new UnsupportedOperationException( + "This ArrayData implementation does not support Geography type."); + } + // ------------------------------------------------------------------------------------------ // Conversion Utilities // ------------------------------------------------------------------------------------------ @@ -225,6 +231,9 @@ static ElementGetter createElementGetter(LogicalType elementType) { case BITMAP: elementGetter = ArrayData::getBitmap; break; + case GEOGRAPHY: + elementGetter = ArrayData::getGeography; + break; case NULL: case SYMBOL: case UNRESOLVED: diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericArrayData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericArrayData.java index f609c2f22f1311..08856cdb0ceb1e 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericArrayData.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericArrayData.java @@ -265,6 +265,11 @@ public Bitmap getBitmap(int pos) { return (Bitmap) getObject(pos); } + @Override + public GeographyData getGeography(int pos) { + return (GeographyData) getObject(pos); + } + private Object getObject(int pos) { return ((Object[]) array)[pos]; } diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericRowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericRowData.java index b234edef62d05b..98c77bca28f06d 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericRowData.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericRowData.java @@ -192,6 +192,11 @@ public byte[] getBinary(int pos) { return (byte[]) this.fields[pos]; } + @Override + public GeographyData getGeography(int pos) { + return (GeographyData) this.fields[pos]; + } + @Override public ArrayData getArray(int pos) { return (ArrayData) this.fields[pos]; diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GeographyData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GeographyData.java new file mode 100644 index 00000000000000..f65075be637c80 --- /dev/null +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GeographyData.java @@ -0,0 +1,82 @@ +/* + * 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.flink.table.data; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.data.binary.BinaryGeographyData; +import org.apache.flink.table.types.logical.GeographyType; + +/** An internal data structure representing data of {@link GeographyType}. */ +@PublicEvolving +public interface GeographyData { + + /** ISO WKB subtype ID for Point geometries. */ + int POINT = 1; + + /** ISO WKB subtype ID for LineString geometries. */ + int LINE_STRING = 2; + + /** ISO WKB subtype ID for Polygon geometries. */ + int POLYGON = 3; + + /** ISO WKB subtype ID for MultiPoint geometries. */ + int MULTI_POINT = 4; + + /** ISO WKB subtype ID for MultiLineString geometries. */ + int MULTI_LINE_STRING = 5; + + /** ISO WKB subtype ID for MultiPolygon geometries. */ + int MULTI_POLYGON = 6; + + /** ISO WKB subtype ID for GeometryCollection geometries. */ + int GEOMETRY_COLLECTION = 7; + + /** + * Converts this {@link GeographyData} object to an ISO WKB byte array. + * + *

Note: The returned byte array may be reused. + */ + byte[] toBytes(); + + /** Returns the ISO WKB subtype ID. */ + int subtypeId(); + + /** Returns the size in bytes of the ISO WKB payload. */ + int sizeInBytes(); + + // ------------------------------------------------------------------------------------------ + // Construction Utilities + // ------------------------------------------------------------------------------------------ + + /** + * Creates an instance of {@link GeographyData} from the given ISO WKB byte array. Returns + * {@code null} if the input is {@code null}. + */ + static GeographyData fromBytes(byte[] bytes) { + return BinaryGeographyData.fromBytes(bytes); + } + + /** + * Creates an instance of {@link GeographyData} from the given ISO WKB byte range. Returns + * {@code null} if the input is {@code null}. + */ + static GeographyData fromBytes(byte[] bytes, int offset, int numBytes) { + return BinaryGeographyData.fromBytes(bytes, offset, numBytes); + } +} diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/RowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/RowData.java index ca43f1608f99ef..a783ecb572a698 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/RowData.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/RowData.java @@ -111,6 +111,8 @@ * +--------------------------------+-----------------------------------------+ * | BITMAP | {@link Bitmap} | * +--------------------------------+-----------------------------------------+ + * | GEOGRAPHY | {@link GeographyData} | + * +--------------------------------+-----------------------------------------+ * * *

Nullability is always handled by the container data structure. @@ -214,6 +216,12 @@ default Bitmap getBitmap(int pos) { "This RowData implementation does not support Bitmap type."); } + /** Returns the geography value at the given position. */ + default GeographyData getGeography(int pos) { + throw new UnsupportedOperationException( + "This RowData implementation does not support Geography type."); + } + // ------------------------------------------------------------------------------------------ // Access Utilities // ------------------------------------------------------------------------------------------ @@ -299,6 +307,9 @@ static FieldGetter createFieldGetter(LogicalType fieldType, int fieldPos) { case BITMAP: fieldGetter = row -> row.getBitmap(fieldPos); break; + case GEOGRAPHY: + fieldGetter = row -> row.getGeography(fieldPos); + break; case NULL: case SYMBOL: case UNRESOLVED: diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java index 10a8b3e6ef71c7..f09f07d25996c1 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java @@ -23,6 +23,7 @@ import org.apache.flink.core.memory.MemorySegmentFactory; import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GeographyData; import org.apache.flink.table.data.MapData; import org.apache.flink.table.data.RawValueData; import org.apache.flink.table.data.RowData; @@ -95,6 +96,7 @@ public static int calculateFixLengthPartSize(LogicalType type) { case RAW: case VARIANT: case BITMAP: + case GEOGRAPHY: // long and double are 8 bytes; // otherwise it stores the length and offset of the variable-length part for types // such as is string, map, etc. @@ -269,6 +271,14 @@ public byte[] getBinary(int pos) { return BinarySegmentUtils.readBinary(segments, offset, fieldOffset, offsetAndSize); } + @Override + public GeographyData getGeography(int pos) { + assertIndexIsValid(pos); + int fieldOffset = getElementOffset(pos, 8); + final long offsetAndSize = BinarySegmentUtils.getLong(segments, fieldOffset); + return BinarySegmentUtils.readGeographyData(segments, offset, fieldOffset, offsetAndSize); + } + @Override public ArrayData getArray(int pos) { assertIndexIsValid(pos); diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryGeographyData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryGeographyData.java new file mode 100644 index 00000000000000..4256f786a86816 --- /dev/null +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryGeographyData.java @@ -0,0 +1,269 @@ +/* + * 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.flink.table.data.binary; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.core.memory.MemorySegment; +import org.apache.flink.core.memory.MemorySegmentFactory; +import org.apache.flink.table.api.TableRuntimeException; +import org.apache.flink.table.data.GeographyData; + +import java.util.Arrays; + +/** + * A binary implementation of {@link GeographyData} backed by raw ISO WKB bytes. + * + *

GEOGRAPHY uses OGC:CRS84 by contract, but ISO WKB does not encode CRS or SRID metadata. This + * container stores the raw ISO WKB payload only; CRS validation, CRS transformation, and EWKB/SRID + * handling belong to constructors, functions, and connector schema mapping. + */ +@Internal +public final class BinaryGeographyData extends BinarySection implements GeographyData { + + private static final int MIN_WKB_HEADER_SIZE = 5; + private static final int WKB_COUNT_SIZE = 4; + private static final int WKB_POINT_COORDINATE_SIZE = 16; + private static final int BIG_ENDIAN = 0; + private static final int LITTLE_ENDIAN = 1; + + private final int subtypeId; + + private BinaryGeographyData(MemorySegment[] segments, int offset, int sizeInBytes) { + super(segments, offset, sizeInBytes); + this.subtypeId = readSubtypeId(segments, offset, sizeInBytes); + } + + /** Creates a {@link BinaryGeographyData} instance from the given address and length. */ + public static BinaryGeographyData fromAddress( + MemorySegment[] segments, int offset, int numBytes) { + return new BinaryGeographyData(segments, offset, numBytes); + } + + /** Creates a {@link BinaryGeographyData} instance from the given ISO WKB bytes. */ + public static BinaryGeographyData fromBytes(byte[] bytes) { + return bytes == null ? null : fromBytes(bytes, 0, bytes.length); + } + + /** Creates a {@link BinaryGeographyData} instance from the given ISO WKB byte range. */ + public static BinaryGeographyData fromBytes(byte[] bytes, int offset, int numBytes) { + if (bytes == null) { + return null; + } + checkRange(bytes, offset, numBytes); + byte[] copy = Arrays.copyOfRange(bytes, offset, offset + numBytes); + return fromAddress(new MemorySegment[] {MemorySegmentFactory.wrap(copy)}, 0, copy.length); + } + + @Override + public byte[] toBytes() { + return BinarySegmentUtils.copyToBytes(segments, offset, sizeInBytes); + } + + @Override + public int subtypeId() { + return subtypeId; + } + + @Override + public int sizeInBytes() { + return sizeInBytes; + } + + private static void checkRange(byte[] bytes, int offset, int numBytes) { + if (offset < 0 || numBytes < 0 || offset > bytes.length - numBytes) { + throw new TableRuntimeException( + String.format( + "Invalid ISO WKB byte range: offset %d, length %d, array length %d.", + offset, numBytes, bytes.length)); + } + } + + private static int readSubtypeId(MemorySegment[] segments, int offset, int sizeInBytes) { + final long endOffset = (long) offset + sizeInBytes; + final GeometryHeader header = readHeader(segments, offset, endOffset); + final long consumedOffset = validateGeometry(segments, offset, endOffset); + if (consumedOffset != endOffset) { + throw new TableRuntimeException( + String.format( + "Malformed ISO WKB payload. Found %d trailing byte(s).", + endOffset - consumedOffset)); + } + return header.subtypeId; + } + + private static long validateGeometry( + MemorySegment[] segments, long geometryOffset, long endOffset) { + final GeometryHeader header = readHeader(segments, geometryOffset, endOffset); + long cursor = geometryOffset + MIN_WKB_HEADER_SIZE; + + switch (header.subtypeId) { + case GeographyData.POINT: + return requireBytes( + cursor, WKB_POINT_COORDINATE_SIZE, endOffset, "POINT coordinates") + + WKB_POINT_COORDINATE_SIZE; + case GeographyData.LINE_STRING: + return skipCoordinateSequence(segments, cursor, endOffset, header.byteOrder); + case GeographyData.POLYGON: + return skipPolygon(segments, cursor, endOffset, header.byteOrder); + case GeographyData.MULTI_POINT: + return skipTypedGeometryCollection( + segments, cursor, endOffset, header.byteOrder, GeographyData.POINT); + case GeographyData.MULTI_LINE_STRING: + return skipTypedGeometryCollection( + segments, cursor, endOffset, header.byteOrder, GeographyData.LINE_STRING); + case GeographyData.MULTI_POLYGON: + return skipTypedGeometryCollection( + segments, cursor, endOffset, header.byteOrder, GeographyData.POLYGON); + case GeographyData.GEOMETRY_COLLECTION: + return skipGeometryCollection(segments, cursor, endOffset, header.byteOrder); + default: + throw new TableRuntimeException( + String.format( + "Malformed ISO WKB payload. Unsupported geography subtype ID %d.", + header.subtypeId)); + } + } + + private static long skipCoordinateSequence( + MemorySegment[] segments, long offset, long endOffset, int byteOrder) { + final long numPoints = + readUnsignedInt(segments, offset, endOffset, byteOrder, "point count"); + long cursor = offset + WKB_COUNT_SIZE; + final long coordinateSequenceSize = + multiplyExact(numPoints, WKB_POINT_COORDINATE_SIZE, "coordinate sequence"); + return requireBytes(cursor, coordinateSequenceSize, endOffset, "coordinate sequence") + + coordinateSequenceSize; + } + + private static long skipPolygon( + MemorySegment[] segments, long offset, long endOffset, int byteOrder) { + final long numRings = readUnsignedInt(segments, offset, endOffset, byteOrder, "ring count"); + long cursor = offset + WKB_COUNT_SIZE; + for (long i = 0; i < numRings; i++) { + cursor = skipCoordinateSequence(segments, cursor, endOffset, byteOrder); + } + return cursor; + } + + private static long skipTypedGeometryCollection( + MemorySegment[] segments, + long offset, + long endOffset, + int byteOrder, + int expectedSubtypeId) { + final long numGeometries = + readUnsignedInt(segments, offset, endOffset, byteOrder, "geometry count"); + long cursor = offset + WKB_COUNT_SIZE; + for (long i = 0; i < numGeometries; i++) { + final GeometryHeader nestedHeader = readHeader(segments, cursor, endOffset); + if (nestedHeader.subtypeId != expectedSubtypeId) { + throw new TableRuntimeException( + String.format( + "Malformed ISO WKB payload. Expected nested subtype ID %d but found %d.", + expectedSubtypeId, nestedHeader.subtypeId)); + } + cursor = validateGeometry(segments, cursor, endOffset); + } + return cursor; + } + + private static long skipGeometryCollection( + MemorySegment[] segments, long offset, long endOffset, int byteOrder) { + final long numGeometries = + readUnsignedInt(segments, offset, endOffset, byteOrder, "geometry count"); + long cursor = offset + WKB_COUNT_SIZE; + for (long i = 0; i < numGeometries; i++) { + cursor = validateGeometry(segments, cursor, endOffset); + } + return cursor; + } + + private static GeometryHeader readHeader( + MemorySegment[] segments, long offset, long endOffset) { + requireBytes(offset, MIN_WKB_HEADER_SIZE, endOffset, "WKB header"); + + final int byteOrder = BinarySegmentUtils.getByte(segments, (int) offset) & 0xFF; + if (byteOrder != BIG_ENDIAN && byteOrder != LITTLE_ENDIAN) { + throw new TableRuntimeException( + String.format( + "Malformed ISO WKB payload. Unsupported byte order %d.", byteOrder)); + } + + final long subtypeId = + readUnsignedInt(segments, offset + 1, endOffset, byteOrder, "subtype ID"); + + if (subtypeId < GeographyData.POINT || subtypeId > GeographyData.GEOMETRY_COLLECTION) { + throw new TableRuntimeException( + String.format( + "Malformed ISO WKB payload. Unsupported geography subtype ID %d.", + subtypeId)); + } + return new GeometryHeader(byteOrder, (int) subtypeId); + } + + private static long readUnsignedInt( + MemorySegment[] segments, + long offset, + long endOffset, + int byteOrder, + String fieldName) { + requireBytes(offset, WKB_COUNT_SIZE, endOffset, fieldName); + + if (byteOrder == LITTLE_ENDIAN) { + return (BinarySegmentUtils.getByte(segments, (int) offset) & 0xFFL) + | ((BinarySegmentUtils.getByte(segments, (int) offset + 1) & 0xFFL) << 8) + | ((BinarySegmentUtils.getByte(segments, (int) offset + 2) & 0xFFL) << 16) + | ((BinarySegmentUtils.getByte(segments, (int) offset + 3) & 0xFFL) << 24); + } + return ((BinarySegmentUtils.getByte(segments, (int) offset) & 0xFFL) << 24) + | ((BinarySegmentUtils.getByte(segments, (int) offset + 1) & 0xFFL) << 16) + | ((BinarySegmentUtils.getByte(segments, (int) offset + 2) & 0xFFL) << 8) + | (BinarySegmentUtils.getByte(segments, (int) offset + 3) & 0xFFL); + } + + private static long requireBytes( + long offset, long numBytes, long endOffset, String componentName) { + if (offset > endOffset || endOffset - offset < numBytes) { + throw new TableRuntimeException( + String.format( + "Malformed ISO WKB payload. Incomplete %s: expected %d byte(s) but found %d.", + componentName, numBytes, Math.max(0, endOffset - offset))); + } + return offset; + } + + private static long multiplyExact(long value, int factor, String componentName) { + final long result = value * factor; + if (value != 0 && result / value != factor) { + throw new TableRuntimeException( + String.format("Malformed ISO WKB payload. %s size overflows.", componentName)); + } + return result; + } + + private static final class GeometryHeader { + private final int byteOrder; + private final int subtypeId; + + private GeometryHeader(int byteOrder, int subtypeId) { + this.byteOrder = byteOrder; + this.subtypeId = subtypeId; + } + } +} diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryRowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryRowData.java index 45bcf7cd3f1128..04c9a0e50ed43d 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryRowData.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryRowData.java @@ -22,6 +22,7 @@ import org.apache.flink.core.memory.MemorySegmentFactory; import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GeographyData; import org.apache.flink.table.data.MapData; import org.apache.flink.table.data.RawValueData; import org.apache.flink.table.data.RowData; @@ -372,6 +373,14 @@ public byte[] getBinary(int pos) { return BinarySegmentUtils.readBinary(segments, offset, fieldOffset, offsetAndLen); } + @Override + public GeographyData getGeography(int pos) { + assertIndexIsValid(pos); + int fieldOffset = getFieldOffset(pos); + final long offsetAndLen = segments[0].getLong(fieldOffset); + return BinarySegmentUtils.readGeographyData(segments, offset, fieldOffset, offsetAndLen); + } + @Override public ArrayData getArray(int pos) { assertIndexIsValid(pos); diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinarySegmentUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinarySegmentUtils.java index 653fc559df9bce..2dd12978d6bf5f 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinarySegmentUtils.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinarySegmentUtils.java @@ -1056,6 +1056,34 @@ public static byte[] readBinary( } } + /** + * Get geography data, if len less than 8, it will be included in variablePartOffsetAndLen. + * + * @param baseOffset base offset of composite binary format. + * @param fieldOffset absolute start offset of 'variablePartOffsetAndLen'. + * @param variablePartOffsetAndLen a long value, real data or offset and len. + */ + public static BinaryGeographyData readGeographyData( + MemorySegment[] segments, + int baseOffset, + int fieldOffset, + long variablePartOffsetAndLen) { + long mark = variablePartOffsetAndLen & HIGHEST_FIRST_BIT; + if (mark == 0) { + final int subOffset = (int) (variablePartOffsetAndLen >> 32); + final int len = (int) variablePartOffsetAndLen; + return BinaryGeographyData.fromAddress(segments, baseOffset + subOffset, len); + } else { + int len = (int) ((variablePartOffsetAndLen & HIGHEST_SECOND_TO_EIGHTH_BIT) >>> 56); + if (BinarySegmentUtils.LITTLE_ENDIAN) { + return BinaryGeographyData.fromAddress(segments, fieldOffset, len); + } else { + // fieldOffset + 1 to skip header. + return BinaryGeographyData.fromAddress(segments, fieldOffset + 1, len); + } + } + } + /** * Get binary string, if len less than 8, will be include in variablePartOffsetAndLen. * diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/NestedRowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/NestedRowData.java index 27085b487ae50d..225d3e103fe103 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/NestedRowData.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/NestedRowData.java @@ -21,6 +21,7 @@ import org.apache.flink.core.memory.MemorySegmentFactory; import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GeographyData; import org.apache.flink.table.data.MapData; import org.apache.flink.table.data.RawValueData; import org.apache.flink.table.data.RowData; @@ -294,6 +295,14 @@ public byte[] getBinary(int pos) { return BinarySegmentUtils.readBinary(segments, offset, fieldOffset, offsetAndLen); } + @Override + public GeographyData getGeography(int pos) { + assertIndexIsValid(pos); + int fieldOffset = getFieldOffset(pos); + final long offsetAndLen = BinarySegmentUtils.getLong(segments, fieldOffset); + return BinarySegmentUtils.readGeographyData(segments, offset, fieldOffset, offsetAndLen); + } + @Override public RowData getRow(int pos, int numFields) { assertIndexIsValid(pos); diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarArrayData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarArrayData.java index a3b8e7dd56daf9..9c6e6ad8c317d0 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarArrayData.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarArrayData.java @@ -21,6 +21,7 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GeographyData; import org.apache.flink.table.data.MapData; import org.apache.flink.table.data.RawValueData; import org.apache.flink.table.data.RowData; @@ -147,6 +148,12 @@ public byte[] getBinary(int pos) { } } + @Override + public GeographyData getGeography(int pos) { + BytesColumnVector.Bytes byteArray = getByteArray(pos); + return GeographyData.fromBytes(byteArray.data, byteArray.offset, byteArray.len); + } + @Override public ArrayData getArray(int pos) { return ((ArrayColumnVector) data).getArray(offset + pos); diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarRowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarRowData.java index bd3dce1edf6bae..0413ac980f737a 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarRowData.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarRowData.java @@ -21,6 +21,7 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GeographyData; import org.apache.flink.table.data.MapData; import org.apache.flink.table.data.RawValueData; import org.apache.flink.table.data.RowData; @@ -152,6 +153,12 @@ public byte[] getBinary(int pos) { } } + @Override + public GeographyData getGeography(int pos) { + Bytes byteArray = vectorizedColumnBatch.getByteArray(rowId, pos); + return GeographyData.fromBytes(byteArray.data, byteArray.offset, byteArray.len); + } + @Override public RowData getRow(int pos, int numFields) { return vectorizedColumnBatch.getRow(rowId, pos); diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/JoinedRowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/JoinedRowData.java index b9eff05a6c388f..d1420debfdb07a 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/JoinedRowData.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/JoinedRowData.java @@ -21,6 +21,7 @@ import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GeographyData; import org.apache.flink.table.data.MapData; import org.apache.flink.table.data.RawValueData; import org.apache.flink.table.data.RowData; @@ -269,6 +270,15 @@ public Bitmap getBitmap(int pos) { } } + @Override + public GeographyData getGeography(int pos) { + if (pos < row1.getArity()) { + return row1.getGeography(pos); + } else { + return row2.getGeography(pos - row1.getArity()); + } + } + @Override public boolean equals(Object o) { if (this == o) { diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ProjectedRowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ProjectedRowData.java index 7f29ca5e613236..e6a173bb4cd991 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ProjectedRowData.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ProjectedRowData.java @@ -22,6 +22,7 @@ import org.apache.flink.table.connector.Projection; import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GeographyData; import org.apache.flink.table.data.MapData; import org.apache.flink.table.data.RawValueData; import org.apache.flink.table.data.RowData; @@ -178,6 +179,11 @@ public Bitmap getBitmap(int pos) { return row.getBitmap(indexMapping[pos]); } + @Override + public GeographyData getGeography(int pos) { + return row.getGeography(indexMapping[pos]); + } + @Override public boolean equals(Object o) { throw new UnsupportedOperationException("Projected row data cannot be compared"); diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/GeographyType.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/GeographyType.java new file mode 100644 index 00000000000000..403ee88e0cfccd --- /dev/null +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/GeographyType.java @@ -0,0 +1,83 @@ +/* + * 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.flink.table.types.logical; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.data.GeographyData; + +import java.util.Collections; +import java.util.List; + +/** + * Data type of geography data. + * + *

The serializable string representation of this type is {@code GEOGRAPHY}. + */ +@PublicEvolving +public final class GeographyType extends LogicalType { + + private static final long serialVersionUID = 1L; + + private static final String FORMAT = "GEOGRAPHY"; + + private static final Class INPUT_OUTPUT_CONVERSION = GeographyData.class; + + public GeographyType(boolean isNullable) { + super(isNullable, LogicalTypeRoot.GEOGRAPHY); + } + + public GeographyType() { + this(true); + } + + @Override + public LogicalType copy(boolean isNullable) { + return new GeographyType(isNullable); + } + + @Override + public String asSerializableString() { + return withNullability(FORMAT); + } + + @Override + public boolean supportsInputConversion(Class clazz) { + return INPUT_OUTPUT_CONVERSION.isAssignableFrom(clazz); + } + + @Override + public boolean supportsOutputConversion(Class clazz) { + return INPUT_OUTPUT_CONVERSION.isAssignableFrom(clazz); + } + + @Override + public Class getDefaultConversion() { + return INPUT_OUTPUT_CONVERSION; + } + + @Override + public List getChildren() { + return Collections.emptyList(); + } + + @Override + public R accept(LogicalTypeVisitor visitor) { + return visitor.visit(this); + } +} diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeRoot.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeRoot.java index 6c823add433bfb..c21abce4af7461 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeRoot.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeRoot.java @@ -145,7 +145,9 @@ public enum LogicalTypeRoot { VARIANT(LogicalTypeFamily.EXTENSION), - BITMAP(LogicalTypeFamily.EXTENSION); + BITMAP(LogicalTypeFamily.EXTENSION), + + GEOGRAPHY(LogicalTypeFamily.EXTENSION); private final Set families; diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java index 6a0e5614466d11..b35ffe05541f44 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java @@ -101,5 +101,9 @@ default R visit(BitmapType bitmapType) { return visit((LogicalType) bitmapType); } + default R visit(GeographyType geographyType) { + return visit((LogicalType) geographyType); + } + R visit(LogicalType other); } diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeDefaultVisitor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeDefaultVisitor.java index 056ec5953ccdc3..43237c443f56e9 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeDefaultVisitor.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeDefaultVisitor.java @@ -31,6 +31,7 @@ import org.apache.flink.table.types.logical.DistinctType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.LogicalType; @@ -203,6 +204,11 @@ public R visit(BitmapType bitmapType) { return defaultMethod(bitmapType); } + @Override + public R visit(GeographyType geographyType) { + return defaultMethod(geographyType); + } + @Override public R visit(LogicalType other) { return defaultMethod(other); diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeParser.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeParser.java index 3711ac76f4fd2a..dfa2e7cf358147 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeParser.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeParser.java @@ -36,6 +36,7 @@ import org.apache.flink.table.types.logical.DescriptorType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LegacyTypeInformationType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; @@ -336,7 +337,8 @@ private enum Keyword { DESCRIPTOR, STRUCTURED, VARIANT, - BITMAP + BITMAP, + GEOGRAPHY } private static final Set KEYWORDS = @@ -586,6 +588,8 @@ private LogicalType parseTypeByKeyword() { return new VariantType(); case BITMAP: return new BitmapType(); + case GEOGRAPHY: + return new GeographyType(); default: throw parsingError("Unsupported type: " + token().value); } diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeUtils.java index 98629b15fc1770..fa994ed160f3c7 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeUtils.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeUtils.java @@ -21,6 +21,7 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GeographyData; import org.apache.flink.table.data.MapData; import org.apache.flink.table.data.RawValueData; import org.apache.flink.table.data.RowData; @@ -115,6 +116,8 @@ public static Class toInternalConversionClass(LogicalType type) { return Variant.class; case BITMAP: return Bitmap.class; + case GEOGRAPHY: + return GeographyData.class; case SYMBOL: case UNRESOLVED: default: diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/LogicalTypeDataTypeConverter.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/LogicalTypeDataTypeConverter.java index 9aa83f3cbee1d3..929aaec786f490 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/LogicalTypeDataTypeConverter.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/LogicalTypeDataTypeConverter.java @@ -38,6 +38,7 @@ import org.apache.flink.table.types.logical.DistinctType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.LogicalType; @@ -265,6 +266,11 @@ public DataType visit(BitmapType bitmapType) { return new AtomicDataType(bitmapType); } + @Override + public DataType visit(GeographyType geographyType) { + return new AtomicDataType(geographyType); + } + @Override public DataType visit(LogicalType other) { if (other.is(LogicalTypeRoot.UNRESOLVED)) { diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/api/SchemaTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/api/SchemaTest.java index 7b54e40f387935..7b4aef14cc3a80 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/api/SchemaTest.java +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/api/SchemaTest.java @@ -51,5 +51,21 @@ void testFromResolvedSchema() { assertThat(newSchema.resolve(new TestSchemaResolver())).isEqualTo(originalSchema); } + + @Test + void testGeographyColumn() { + final Schema schema = + Schema.newBuilder() + .column("location", DataTypes.GEOGRAPHY()) + .column("required_location", DataTypes.GEOGRAPHY().notNull()) + .build(); + + assertThat(schema.resolve(new TestSchemaResolver())) + .isEqualTo( + ResolvedSchema.of( + Column.physical("location", DataTypes.GEOGRAPHY()), + Column.physical( + "required_location", DataTypes.GEOGRAPHY().notNull()))); + } } } diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/BinaryGeographyDataTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/BinaryGeographyDataTest.java new file mode 100644 index 00000000000000..4e45d2d28549aa --- /dev/null +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/BinaryGeographyDataTest.java @@ -0,0 +1,331 @@ +/* + * 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.flink.table.data.binary; + +import org.apache.flink.core.memory.MemorySegment; +import org.apache.flink.core.memory.MemorySegmentFactory; +import org.apache.flink.table.api.TableRuntimeException; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.GeographyData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.GeographyType; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.util.Arrays; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link BinaryGeographyData}. */ +class BinaryGeographyDataTest { + + private static final int BIG_ENDIAN = 0; + private static final int LITTLE_ENDIAN = 1; + private static final byte[] POINT_WKB = pointWkb(LITTLE_ENDIAN); + + @Test + void testValidWkbRoundTrip() { + final GeographyData geography = GeographyData.fromBytes(POINT_WKB); + + assertThat(geography).isInstanceOf(BinaryGeographyData.class); + assertThat(geography.toBytes()).isEqualTo(POINT_WKB); + assertThat(geography.subtypeId()).isEqualTo(GeographyData.POINT); + assertThat(geography.sizeInBytes()).isEqualTo(POINT_WKB.length); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("supportedSubtypePayloads") + void testSupported2dSubtypePayloads(String name, int subtypeId, byte[] wkb) { + final BinaryGeographyData geography = BinaryGeographyData.fromBytes(wkb); + + assertThat(geography.subtypeId()).isEqualTo(subtypeId); + assertThat(geography.toBytes()).isEqualTo(wkb); + assertThat(geography.sizeInBytes()).isEqualTo(wkb.length); + } + + static Stream supportedSubtypePayloads() { + return Stream.of( + Arguments.of("Point", GeographyData.POINT, pointWkb(LITTLE_ENDIAN)), + Arguments.of("LineString", GeographyData.LINE_STRING, lineStringWkb(LITTLE_ENDIAN)), + Arguments.of("Polygon", GeographyData.POLYGON, polygonWkb(LITTLE_ENDIAN)), + Arguments.of("MultiPoint", GeographyData.MULTI_POINT, multiPointWkb(LITTLE_ENDIAN)), + Arguments.of( + "MultiLineString", + GeographyData.MULTI_LINE_STRING, + multiLineStringWkb(LITTLE_ENDIAN)), + Arguments.of( + "MultiPolygon", + GeographyData.MULTI_POLYGON, + multiPolygonWkb(LITTLE_ENDIAN)), + Arguments.of( + "GeometryCollection", + GeographyData.GEOMETRY_COLLECTION, + geometryCollectionWkb(LITTLE_ENDIAN))); + } + + @Test + void testBigEndianSubtypeExtraction() { + final BinaryGeographyData point = BinaryGeographyData.fromBytes(pointWkb(BIG_ENDIAN)); + final BinaryGeographyData lineString = + BinaryGeographyData.fromBytes(lineStringWkb(BIG_ENDIAN)); + + assertThat(point.subtypeId()).isEqualTo(GeographyData.POINT); + assertThat(lineString.subtypeId()).isEqualTo(GeographyData.LINE_STRING); + } + + @Test + void testNullHandling() { + assertThat(GeographyData.fromBytes((byte[]) null)).isNull(); + assertThat(GeographyData.fromBytes(null, 0, 0)).isNull(); + assertThat(BinaryGeographyData.fromBytes((byte[]) null)).isNull(); + assertThat(BinaryGeographyData.fromBytes(null, 0, 0)).isNull(); + } + + @Test + void testGenericRowDataNullPath() { + final RowData.FieldGetter getter = RowData.createFieldGetter(new GeographyType(), 0); + final GenericRowData nullRow = GenericRowData.of((Object) null); + final GeographyData geography = GeographyData.fromBytes(POINT_WKB); + final GenericRowData valueRow = GenericRowData.of(geography); + + assertThat(getter.getFieldOrNull(nullRow)).isNull(); + assertThat(getter.getFieldOrNull(valueRow)).isSameAs(geography); + } + + @Test + void testBinaryRowDataNullPath() { + final RowData.FieldGetter getter = RowData.createFieldGetter(new GeographyType(), 0); + final int fixedLength = BinaryRowData.calculateFixPartSizeInBytes(1); + final BinaryRowData nullRow = new BinaryRowData(1); + nullRow.pointTo(MemorySegmentFactory.wrap(new byte[fixedLength]), 0, fixedLength); + nullRow.setNullAt(0); + + assertThat(getter.getFieldOrNull(nullRow)).isNull(); + } + + @Test + void testBinaryRowDataValuePath() { + final BinaryRowData row = binaryRowWithGeography(POINT_WKB); + + assertThat(row.getGeography(0).toBytes()).isEqualTo(POINT_WKB); + assertThat(row.getGeography(0).subtypeId()).isEqualTo(GeographyData.POINT); + } + + @Test + void testByteRangeCopiesOnlySelectedPayload() { + final byte[] bytes = concat(bytes(42), POINT_WKB, bytes(99)); + final BinaryGeographyData geography = + BinaryGeographyData.fromBytes(bytes, 1, POINT_WKB.length); + + bytes[1] = 0; + + assertThat(geography.toBytes()).isEqualTo(POINT_WKB); + assertThat(geography.subtypeId()).isEqualTo(GeographyData.POINT); + } + + @Test + void testFromAddressSupportsSegmentBoundary() { + final byte[] first = concat(new byte[14], Arrays.copyOfRange(POINT_WKB, 0, 2)); + final byte[] second = Arrays.copyOfRange(POINT_WKB, 2, 18); + final byte[] third = + concat(Arrays.copyOfRange(POINT_WKB, 18, POINT_WKB.length), new byte[13]); + final MemorySegment[] segments = + new MemorySegment[] { + MemorySegmentFactory.wrap(first), + MemorySegmentFactory.wrap(second), + MemorySegmentFactory.wrap(third) + }; + + final BinaryGeographyData geography = + BinaryGeographyData.fromAddress(segments, 14, POINT_WKB.length); + + assertThat(geography.toBytes()).isEqualTo(POINT_WKB); + assertThat(geography.subtypeId()).isEqualTo(GeographyData.POINT); + } + + @Test + void testMalformedPayloadBoundaries() { + assertThatThrownBy(() -> BinaryGeographyData.fromBytes(bytes())) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContaining("Incomplete WKB header"); + + assertThatThrownBy(() -> BinaryGeographyData.fromBytes(bytes(1, 1, 0, 0))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContaining("Incomplete WKB header"); + + assertThatThrownBy(() -> BinaryGeographyData.fromBytes(bytes(2, 1, 0, 0, 0))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContaining("Unsupported byte order 2"); + + assertThatThrownBy(() -> BinaryGeographyData.fromBytes(header(LITTLE_ENDIAN, 0))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContaining("Unsupported geography subtype ID 0"); + + assertThatThrownBy(() -> BinaryGeographyData.fromBytes(header(LITTLE_ENDIAN, 8))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContaining("Unsupported geography subtype ID 8"); + } + + @Test + void testStructurallyIncompletePayloads() { + assertThatThrownBy( + () -> + BinaryGeographyData.fromBytes( + header(LITTLE_ENDIAN, GeographyData.POINT))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContaining("Incomplete POINT coordinates"); + + assertThatThrownBy( + () -> + BinaryGeographyData.fromBytes( + concat( + header(LITTLE_ENDIAN, GeographyData.LINE_STRING), + unsignedInt(LITTLE_ENDIAN, 1), + new byte[15]))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContaining("Incomplete coordinate sequence"); + + assertThatThrownBy( + () -> + BinaryGeographyData.fromBytes( + concat( + header(LITTLE_ENDIAN, GeographyData.MULTI_POINT), + unsignedInt(LITTLE_ENDIAN, 1), + lineStringWkb(LITTLE_ENDIAN)))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContaining("Expected nested subtype ID 1 but found 2"); + + assertThatThrownBy( + () -> + BinaryGeographyData.fromBytes( + concat(pointWkb(LITTLE_ENDIAN), bytes(42)))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContaining("trailing byte"); + } + + @Test + void testInvalidByteRanges() { + assertThatThrownBy(() -> BinaryGeographyData.fromBytes(POINT_WKB, -1, POINT_WKB.length)) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContaining("Invalid ISO WKB byte range"); + + assertThatThrownBy(() -> BinaryGeographyData.fromBytes(POINT_WKB, 0, POINT_WKB.length + 1)) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContaining("Invalid ISO WKB byte range"); + } + + private static BinaryRowData binaryRowWithGeography(byte[] wkb) { + final int fixedLength = BinaryRowData.calculateFixPartSizeInBytes(1); + final byte[] rowBytes = new byte[fixedLength + wkb.length]; + final MemorySegment segment = MemorySegmentFactory.wrap(rowBytes); + segment.putLong(8, ((long) fixedLength << 32) | wkb.length); + segment.put(fixedLength, wkb, 0, wkb.length); + + final BinaryRowData row = new BinaryRowData(1); + row.pointTo(segment, 0, rowBytes.length); + return row; + } + + private static byte[] pointWkb(int byteOrder) { + return concat(header(byteOrder, GeographyData.POINT), new byte[16]); + } + + private static byte[] lineStringWkb(int byteOrder) { + return concat( + header(byteOrder, GeographyData.LINE_STRING), + unsignedInt(byteOrder, 2), + new byte[32]); + } + + private static byte[] polygonWkb(int byteOrder) { + return concat( + header(byteOrder, GeographyData.POLYGON), + unsignedInt(byteOrder, 1), + unsignedInt(byteOrder, 4), + new byte[64]); + } + + private static byte[] multiPointWkb(int byteOrder) { + return concat( + header(byteOrder, GeographyData.MULTI_POINT), + unsignedInt(byteOrder, 1), + pointWkb(byteOrder)); + } + + private static byte[] multiLineStringWkb(int byteOrder) { + return concat( + header(byteOrder, GeographyData.MULTI_LINE_STRING), + unsignedInt(byteOrder, 1), + lineStringWkb(byteOrder)); + } + + private static byte[] multiPolygonWkb(int byteOrder) { + return concat( + header(byteOrder, GeographyData.MULTI_POLYGON), + unsignedInt(byteOrder, 1), + polygonWkb(byteOrder)); + } + + private static byte[] geometryCollectionWkb(int byteOrder) { + return concat( + header(byteOrder, GeographyData.GEOMETRY_COLLECTION), + unsignedInt(byteOrder, 2), + pointWkb(byteOrder), + lineStringWkb(byteOrder)); + } + + private static byte[] header(int byteOrder, int subtypeId) { + return concat(bytes(byteOrder), unsignedInt(byteOrder, subtypeId)); + } + + private static byte[] unsignedInt(int byteOrder, int value) { + if (byteOrder == LITTLE_ENDIAN) { + return bytes(value, value >>> 8, value >>> 16, value >>> 24); + } + return bytes(value >>> 24, value >>> 16, value >>> 8, value); + } + + private static byte[] concat(byte[]... values) { + int length = 0; + for (byte[] value : values) { + length += value.length; + } + + final byte[] result = new byte[length]; + int offset = 0; + for (byte[] value : values) { + System.arraycopy(value, 0, result, offset, value.length); + offset += value.length; + } + return result; + } + + private static byte[] bytes(int... values) { + final byte[] result = new byte[values.length]; + for (int i = 0; i < values.length; i++) { + result[i] = (byte) values[i]; + } + return result; + } +} diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/DataTypesTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/DataTypesTest.java index 8c274ed719137e..3c0bf063631459 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/DataTypesTest.java +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/DataTypesTest.java @@ -21,6 +21,7 @@ import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.common.typeutils.base.VoidSerializer; import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.data.GeographyData; import org.apache.flink.table.types.logical.ArrayType; import org.apache.flink.table.types.logical.BigIntType; import org.apache.flink.table.types.logical.BinaryType; @@ -32,6 +33,7 @@ import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.LogicalType; @@ -82,6 +84,7 @@ import static org.apache.flink.table.api.DataTypes.DOUBLE; import static org.apache.flink.table.api.DataTypes.FIELD; import static org.apache.flink.table.api.DataTypes.FLOAT; +import static org.apache.flink.table.api.DataTypes.GEOGRAPHY; import static org.apache.flink.table.api.DataTypes.INT; import static org.apache.flink.table.api.DataTypes.INTERVAL; import static org.apache.flink.table.api.DataTypes.MAP; @@ -231,6 +234,9 @@ private static Stream testData() { TestSpec.forDataType(BITMAP()) .expectLogicalType(new BitmapType()) .expectConversionClass(Bitmap.class), + TestSpec.forDataType(GEOGRAPHY()) + .expectLogicalType(new GeographyType()) + .expectConversionClass(GeographyData.class), TestSpec.forUnresolvedDataType(RAW(Types.VOID)) .expectUnresolvedString("[RAW('java.lang.Void', '?')]") .lookupReturns(dummyRaw(Void.class)) diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java index 6fb3574b1f59f6..89cc1341dfe2a8 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java @@ -31,6 +31,7 @@ import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.LogicalType; @@ -295,7 +296,13 @@ private static Stream testData() { new VariantType(), new MapType(new IntType(), new CharType()), false, - false)); + false), + + // GEOGRAPHY construction and serialization require explicit functions. + Arguments.of(new GeographyType(), VarCharType.STRING_TYPE, false, false), + Arguments.of(VarCharType.STRING_TYPE, new GeographyType(), false, false), + Arguments.of(new GeographyType(), new VarBinaryType(), false, false), + Arguments.of(new VarBinaryType(), new GeographyType(), false, false)); } @ParameterizedTest(name = "{index}: [From: {0}, To: {1}, Implicit: {2}, Explicit: {3}]") diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeParserTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeParserTest.java index 258b397c871878..e1c96b1fbc9e7c 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeParserTest.java +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeParserTest.java @@ -39,6 +39,7 @@ import org.apache.flink.table.types.logical.DescriptorType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LegacyTypeInformationType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; @@ -309,6 +310,8 @@ private static Stream testData() { TestSpec.forString("VARIANT NOT NULL").expectType(new VariantType(false)), TestSpec.forString("BITMAP").expectType(new BitmapType()), TestSpec.forString("BITMAP NOT NULL").expectType(new BitmapType(false)), + TestSpec.forString("GEOGRAPHY").expectType(new GeographyType()), + TestSpec.forString("GEOGRAPHY NOT NULL").expectType(new GeographyType(false)), // error message testing diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java index ee40463cf0b92d..b608d923cc3b6a 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java @@ -26,6 +26,8 @@ import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.catalog.ObjectIdentifier; import org.apache.flink.table.catalog.UnresolvedIdentifier; +import org.apache.flink.table.data.GeographyData; +import org.apache.flink.table.data.binary.BinaryGeographyData; import org.apache.flink.table.expressions.TimeIntervalUnit; import org.apache.flink.table.legacy.types.logical.TypeInformationRawType; import org.apache.flink.table.types.logical.ArrayType; @@ -40,6 +42,7 @@ import org.apache.flink.table.types.logical.DistinctType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.LogicalType; @@ -629,6 +632,20 @@ void testBitmapType() { new BitmapType(false))); } + @Test + void testGeographyType() { + assertThat(new GeographyType()) + .isJavaSerializable() + .satisfies( + baseAssertions( + "GEOGRAPHY", + "GEOGRAPHY", + new Class[] {GeographyData.class, BinaryGeographyData.class}, + new Class[] {GeographyData.class, BinaryGeographyData.class}, + new LogicalType[] {}, + new GeographyType(false))); + } + @Test void testTypeInformationRawType() { final TypeInformationRawType rawType = diff --git a/flink-table/flink-table-planner/pom.xml b/flink-table/flink-table-planner/pom.xml index b8f76f7f969929..b1686bf1f0d28a 100644 --- a/flink-table/flink-table-planner/pom.xml +++ b/flink-table/flink-table-planner/pom.xml @@ -158,6 +158,11 @@ under the License. flink-table-runtime ${project.version} + + org.apache.flink + flink-table-type-utils + ${project.version} + @@ -183,6 +188,13 @@ under the License. test + + org.apache.flink + flink-table-code-splitter + ${project.version} + test + + org.apache.flink diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlCastFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlCastFunction.java index 7b6112f9672dc8..d819a145e94cdd 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlCastFunction.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlCastFunction.java @@ -269,16 +269,22 @@ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFail private boolean canCastFrom(RelDataType toType, RelDataType fromType) { SqlTypeName fromTypeName = fromType.getSqlTypeName(); + SqlTypeName toTypeName = toType.getSqlTypeName(); // Cast to Variant is not support at the moment. // TODO: Support cast to variant (FLINK-37925，FLINK-37926) - if (toType.getSqlTypeName() == SqlTypeName.VARIANT) { + if (toTypeName == SqlTypeName.VARIANT) { return false; } // Cast to BITMAP is not supported at the moment. if (toType instanceof BitmapRelDataType) { return false; } + if (toTypeName == SqlTypeName.OTHER) { + return LogicalTypeCasts.supportsExplicitCast( + FlinkTypeFactory.toLogicalType(fromType), + FlinkTypeFactory.toLogicalType(toType)); + } switch (fromTypeName) { case ARRAY: case MAP: diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkTypeFactory.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkTypeFactory.java index bb99230775c855..703c5707583975 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkTypeFactory.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkTypeFactory.java @@ -29,6 +29,7 @@ import org.apache.flink.table.legacy.types.logical.TypeInformationRawType; import org.apache.flink.table.planner.plan.schema.BitmapRelDataType; import org.apache.flink.table.planner.plan.schema.GenericRelDataType; +import org.apache.flink.table.planner.plan.schema.GeographyRelDataType; import org.apache.flink.table.planner.plan.schema.RawRelDataType; import org.apache.flink.table.planner.plan.schema.StructuredRelDataType; import org.apache.flink.table.planner.plan.schema.TimeIndicatorRelDataType; @@ -46,6 +47,7 @@ import org.apache.flink.table.types.logical.DescriptorType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LegacyTypeInformationType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; @@ -65,6 +67,7 @@ import org.apache.flink.table.types.logical.VarBinaryType; import org.apache.flink.table.types.logical.VarCharType; import org.apache.flink.table.types.logical.VariantType; +import org.apache.flink.table.types.logical.utils.LogicalTypeMerging; import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo; import org.apache.flink.table.utils.TableSchemaUtils; import org.apache.flink.util.Preconditions; @@ -160,6 +163,11 @@ public RelDataType createBitmapType() { return canonize(new BitmapRelDataType(new BitmapType())); } + @Override + public RelDataType createGeographyType() { + return canonize(new GeographyRelDataType(new GeographyType())); + } + @Override public RelDataType createArrayType(RelDataType elementType, long maxCardinality) { // Just validate type, make sure there is a failure in validate phase. @@ -230,6 +238,8 @@ public RelDataType createTypeWithNullability(RelDataType relDataType, boolean is newType = ((StructuredRelDataType) relDataType).createWithNullability(isNullable); } else if (relDataType instanceof BitmapRelDataType) { newType = ((BitmapRelDataType) relDataType).createWithNullability(isNullable); + } else if (relDataType instanceof GeographyRelDataType) { + newType = ((GeographyRelDataType) relDataType).createWithNullability(isNullable); } else if (relDataType instanceof GenericRelDataType) { final GenericRelDataType generic = (GenericRelDataType) relDataType; newType = new GenericRelDataType(generic.genericType(), isNullable, getTypeSystem()); @@ -254,8 +264,20 @@ public RelDataType createTypeWithNullability(RelDataType relDataType, boolean is @Override public RelDataType leastRestrictive(List types) { final Optional resolved = resolveAllIdenticalTypes(types); - final RelDataType leastRestrictive = - resolved.orElseGet(() -> super.leastRestrictive(types)); + if (resolved.isPresent()) { + return normalizeLeastRestrictive(resolved.get()); + } + + if (containsFlinkExtensionType(types)) { + return normalizeLeastRestrictive( + resolveCommonTypeForFlinkExtensions(types).orElse(null)); + } + + final RelDataType leastRestrictive = super.leastRestrictive(types); + return normalizeLeastRestrictive(leastRestrictive); + } + + private RelDataType normalizeLeastRestrictive(RelDataType leastRestrictive) { // NULL is reserved for untyped literals only if (leastRestrictive == null || leastRestrictive.getSqlTypeName() == SqlTypeName.NULL) { return null; @@ -263,6 +285,24 @@ public RelDataType leastRestrictive(List types) { return leastRestrictive; } + private Optional resolveCommonTypeForFlinkExtensions(List types) { + return LogicalTypeMerging.findCommonType( + types.stream() + .map(FlinkTypeFactory::toLogicalType) + .collect(Collectors.toList())) + .map(this::createFieldTypeFromLogicalType); + } + + private boolean containsFlinkExtensionType(List types) { + return types.stream().anyMatch(FlinkTypeFactory::isFlinkExtensionType); + } + + private static boolean isFlinkExtensionType(RelDataType type) { + return type instanceof RawRelDataType + || type instanceof BitmapRelDataType + || type instanceof GeographyRelDataType; + } + private Optional resolveAllIdenticalTypes(List types) { final RelDataType head = types.get(0); // check if all types are the same @@ -492,6 +532,9 @@ private RelDataType newRelDataType(LogicalType logicalType) { case BITMAP: return new BitmapRelDataType((BitmapType) logicalType); + case GEOGRAPHY: + return new GeographyRelDataType((GeographyType) logicalType); + default: throw new TableException("Type is not supported: " + logicalType); } @@ -865,6 +908,8 @@ private static LogicalType toLogicalTypeWithoutNullability(RelDataType relDataTy return ((RawRelDataType) relDataType).getRawType(); } else if (relDataType instanceof BitmapRelDataType) { return ((BitmapRelDataType) relDataType).getBitmapType(); + } else if (relDataType instanceof GeographyRelDataType) { + return ((GeographyRelDataType) relDataType).getGeographyType(); } else { throw new TableException("Type is not supported: " + relDataType); } diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerializer.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerializer.java index 2857c4dbaa3bcb..6429a2c5586a86 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerializer.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerializer.java @@ -544,6 +544,7 @@ protected Boolean defaultMethod(LogicalType logicalType) { case NULL: case DESCRIPTOR: case BITMAP: + case GEOGRAPHY: return true; default: // fall back to generic serialization diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/schema/GeographyRelDataType.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/schema/GeographyRelDataType.java new file mode 100644 index 00000000000000..41b586f5863240 --- /dev/null +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/schema/GeographyRelDataType.java @@ -0,0 +1,62 @@ +/* + * 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.flink.table.planner.plan.schema; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.types.logical.GeographyType; + +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.sql.type.AbstractSqlType; +import org.apache.calcite.sql.type.SqlTypeName; + +/** The {@link RelDataType} representation of a {@link GeographyType}. */ +@Internal +public final class GeographyRelDataType extends AbstractSqlType { + + private final GeographyType geographyType; + + public GeographyRelDataType(GeographyType geographyType) { + super(SqlTypeName.OTHER, geographyType.isNullable(), null); + this.geographyType = geographyType; + computeDigest(); + } + + public GeographyType getGeographyType() { + return geographyType; + } + + public GeographyRelDataType createWithNullability(boolean nullable) { + if (nullable == isNullable()) { + return this; + } + return new GeographyRelDataType((GeographyType) geographyType.copy(nullable)); + } + + @Override + protected void generateTypeString(StringBuilder sb, boolean withDetail) { + sb.append(geographyType.asSummaryString()); + } + + @Override + protected void computeDigest() { + final StringBuilder sb = new StringBuilder(); + generateTypeString(sb, true); + digest = sb.toString(); + } +} diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/LogicalRelDataTypeConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/LogicalRelDataTypeConverter.java index 12093cfe1e5711..145376f781a07b 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/LogicalRelDataTypeConverter.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/LogicalRelDataTypeConverter.java @@ -23,6 +23,7 @@ import org.apache.flink.table.catalog.DataTypeFactory; import org.apache.flink.table.planner.calcite.FlinkTypeFactory; import org.apache.flink.table.planner.plan.schema.BitmapRelDataType; +import org.apache.flink.table.planner.plan.schema.GeographyRelDataType; import org.apache.flink.table.planner.plan.schema.RawRelDataType; import org.apache.flink.table.planner.plan.schema.StructuredRelDataType; import org.apache.flink.table.planner.plan.schema.TimeIndicatorRelDataType; @@ -40,6 +41,7 @@ import org.apache.flink.table.types.logical.DistinctType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.LogicalType; @@ -466,6 +468,11 @@ public RelDataType visit(BitmapType bitmapType) { return new BitmapRelDataType(bitmapType); } + @Override + public RelDataType visit(GeographyType geographyType) { + return new GeographyRelDataType(geographyType); + } + @Override public RelDataType visit(LogicalType other) { throw new TableException( @@ -596,6 +603,8 @@ private static LogicalType toLogicalTypeNotNull( return ((RawRelDataType) relDataType).getRawType(); } else if (relDataType instanceof BitmapRelDataType) { return ((BitmapRelDataType) relDataType).getBitmapType(); + } else if (relDataType instanceof GeographyRelDataType) { + return ((GeographyRelDataType) relDataType).getGeographyType(); } // fall through case REAL: diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala index ff924fe0f30f8f..148d85ad969d2c 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala @@ -279,6 +279,7 @@ object CodeGenUtils { case DESCRIPTOR => className[ColumnList] case VARIANT => className[Variant] case BITMAP => className[Bitmap] + case GEOGRAPHY => className[GeographyData] case SYMBOL | UNRESOLVED => throw new IllegalArgumentException("Illegal type: " + t) } @@ -386,6 +387,8 @@ object CodeGenUtils { s"$term.toObject($serTerm).hashCode()" case BITMAP => s"$term.hashCode()" + case GEOGRAPHY => + s"$term.hashCode()" case NULL | SYMBOL | UNRESOLVED => throw new IllegalArgumentException("Illegal type: " + t) } @@ -538,6 +541,8 @@ object CodeGenUtils { s"$rowTerm.getVariant($indexTerm)" case BITMAP => s"$rowTerm.getBitmap($indexTerm)" + case GEOGRAPHY => + s"$rowTerm.getGeography($indexTerm)" case NULL | SYMBOL | UNRESOLVED => throw new IllegalArgumentException("Illegal type: " + t) } @@ -835,6 +840,8 @@ object CodeGenUtils { s"$writerTerm.writeVariant($indexTerm, $fieldValTerm)" case BITMAP => s"$writerTerm.writeBitmap($indexTerm, $fieldValTerm)" + case GEOGRAPHY => + s"$writerTerm.writeGeography($indexTerm, $fieldValTerm)" case NULL | SYMBOL | UNRESOLVED => throw new IllegalArgumentException("Illegal type: " + t); } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java index 0d5c13d48a538b..aa0039b7bd16bf 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java @@ -86,6 +86,33 @@ void testCompilePlanSql() throws IOException { .isEqualTo(getPreparedToCompareCompiledPlan(expected)); } + @Test + void testCompilePlanSqlWithGeographyColumns() { + tableEnv.executeSql( + "CREATE TABLE GeoSource (" + + "id INT, " + + "location GEOGRAPHY, " + + "required_location GEOGRAPHY NOT NULL" + + ") WITH (" + + "'connector' = 'values', " + + "'bounded' = 'false')"); + tableEnv.executeSql( + "CREATE TABLE GeoSink (" + + "id INT, " + + "location GEOGRAPHY, " + + "required_location GEOGRAPHY NOT NULL" + + ") WITH (" + + "'connector' = 'values', " + + "'table-sink-class' = 'DEFAULT')"); + + final CompiledPlan compiledPlan = + tableEnv.compilePlanSql("INSERT INTO GeoSink SELECT * FROM GeoSource"); + final String planJson = compiledPlan.asJsonString(); + + assertThat(planJson).contains("\"GEOGRAPHY\"", "\"GEOGRAPHY NOT NULL\""); + assertThat(tableEnv.loadPlan(PlanReference.fromJsonString(planJson))).isNotNull(); + } + @Test void testSourceTableWithHints() { CompiledPlan compiledPlan = diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidatorTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidatorTest.java index 831b3b6ecc480a..de08fb2c8e6317 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidatorTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidatorTest.java @@ -26,12 +26,21 @@ import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.functions.ScalarFunction; import org.apache.flink.table.planner.utils.PlannerMocks; +import org.apache.flink.table.types.logical.ArrayType; +import org.apache.flink.table.types.logical.GeographyType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.MapType; +import org.apache.flink.table.types.logical.VarCharType; +import org.apache.calcite.rel.RelRoot; +import org.apache.calcite.sql.SqlNode; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assertions.catchThrowable; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; /** Test for {@link FlinkCalciteSqlValidator}. */ @@ -158,6 +167,53 @@ void testMixedPositionalAndNamedArguments() { .hasMessageContaining("Cannot mix positional and named arguments"); } + @Test + void testArrayConstructorInfersGeographyElementType() { + LogicalType logicalType = + projectedLogicalType( + "ARRAY[ST_GEOGFROMTEXT('POINT (0 0)'), ST_GEOGFROMTEXT('POINT (1 1)')]"); + + assertThat(logicalType).isEqualTo(new ArrayType(new GeographyType())); + } + + @Test + void testArrayConstructorInfersNullableGeographyElementType() { + LogicalType logicalType = + projectedLogicalType("ARRAY[ST_GEOGFROMTEXT('POINT (0 0)'), NULL]"); + + assertThat(logicalType).isEqualTo(new ArrayType(new GeographyType())); + } + + @Test + void testMapConstructorInfersGeographyValueType() { + LogicalType logicalType = + projectedLogicalType( + "MAP['a', ST_GEOGFROMTEXT('POINT (0 0)'), 'b', ST_GEOGFROMTEXT('POINT (1 1)')]"); + + assertThat(logicalType) + .isEqualTo(new MapType(VarCharType.STRING_TYPE, new GeographyType())); + } + + @Test + void testIncompatibleExtensionTypesFailValidation() { + Throwable thrown = + catchThrowable( + () -> + projectedLogicalType( + "ARRAY[ST_GEOGFROMTEXT('POINT (0 0)'), BITMAP_BUILD(1)]")); + + assertThat(thrown).isInstanceOf(ValidationException.class); + assertThat(thrown.getCause()).isNotInstanceOf(AssertionError.class); + } + + private LogicalType projectedLogicalType(String expression) { + SqlNode parsed = plannerMocks.getPlanner().parser().parse("SELECT " + expression + " AS c"); + SqlNode validated = plannerMocks.getPlanner().validate(parsed); + RelRoot relRoot = plannerMocks.getPlanner().rel(validated); + return FlinkTypeFactory.toLogicalType( + relRoot.rel.getRowType().getFieldList().get(0).getType()); + } + /** Scalar function with named arguments for the mixed-argument validation test. */ public static class NamedArgsScalarFunction extends ScalarFunction { @FunctionHint( diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkTypeFactoryTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkTypeFactoryTest.java index 7715f5bacddcbb..ae775d7446c7ba 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkTypeFactoryTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkTypeFactoryTest.java @@ -25,12 +25,14 @@ import org.apache.flink.table.legacy.types.logical.TypeInformationRawType; import org.apache.flink.table.types.logical.ArrayType; import org.apache.flink.table.types.logical.BigIntType; +import org.apache.flink.table.types.logical.BitmapType; import org.apache.flink.table.types.logical.BooleanType; import org.apache.flink.table.types.logical.CharType; import org.apache.flink.table.types.logical.DateType; import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.LogicalType; @@ -221,6 +223,37 @@ void testLeastRestrictive(List input, LogicalType expected) { .isEqualTo(typeFactory.createFieldTypeFromLogicalType(expected)); } + @Test + void testLeastRestrictiveGeographyNullability() { + FlinkTypeFactory typeFactory = + new FlinkTypeFactory( + Thread.currentThread().getContextClassLoader(), FlinkTypeSystem.INSTANCE); + + assertThat( + typeFactory.leastRestrictive( + Stream.of( + new GeographyType(false), + new GeographyType(true), + new NullType()) + .map(typeFactory::createFieldTypeFromLogicalType) + .collect(Collectors.toList()))) + .isEqualTo(typeFactory.createFieldTypeFromLogicalType(new GeographyType(true))); + } + + @Test + void testLeastRestrictiveIncompatibleExtensionTypes() { + FlinkTypeFactory typeFactory = + new FlinkTypeFactory( + Thread.currentThread().getContextClassLoader(), FlinkTypeSystem.INSTANCE); + + assertThat( + typeFactory.leastRestrictive( + Stream.of(new GeographyType(), new BitmapType()) + .map(typeFactory::createFieldTypeFromLogicalType) + .collect(Collectors.toList()))) + .isNull(); + } + public static class TestClass { public int f0; public String f1; diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/ProjectionCodeGeneratorGeographyTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/ProjectionCodeGeneratorGeographyTest.java new file mode 100644 index 00000000000000..652d0af25d6427 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/ProjectionCodeGeneratorGeographyTest.java @@ -0,0 +1,65 @@ +/* + * 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.flink.table.planner.codegen; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.GeographyData; +import org.apache.flink.table.data.binary.BinaryRowData; +import org.apache.flink.table.runtime.generated.Projection; +import org.apache.flink.table.types.logical.GeographyType; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.RowType; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for generated projections involving {@link GeographyData}. */ +class ProjectionCodeGeneratorGeographyTest { + + private static final byte[] POINT_WKB = + new byte[] { + 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, (byte) 0xF0, 0x3F, 0, 0, 0, 0, 0, 0, 0, 0x40 + }; + + @Test + void testGeneratedProjectionForGeography() { + RowType inputType = RowType.of(new IntType(), new GeographyType(), new GeographyType()); + RowType outputType = RowType.of(new GeographyType(), new GeographyType(), new IntType()); + GenericRowData input = GenericRowData.of(7, GeographyData.fromBytes(POINT_WKB), null); + + Projection projection = + ProjectionCodeGenerator.generateProjection( + new CodeGeneratorContext( + new Configuration(), + Thread.currentThread().getContextClassLoader()), + "GeographyProjection", + inputType, + outputType, + new int[] {1, 2, 0}) + .newInstance(Thread.currentThread().getContextClassLoader()); + + BinaryRowData output = (BinaryRowData) projection.apply(input); + + assertThat(output.getGeography(0).toBytes()).isEqualTo(POINT_WKB); + assertThat(output.isNullAt(1)).isTrue(); + assertThat(output.getInt(2)).isEqualTo(7); + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionMiscITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionMiscITCase.java index 218d911e881dba..f37a8acfa4ac47 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionMiscITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionMiscITCase.java @@ -39,6 +39,7 @@ import static org.apache.flink.table.api.DataTypes.BOOLEAN; import static org.apache.flink.table.api.DataTypes.BYTES; import static org.apache.flink.table.api.DataTypes.FIELD; +import static org.apache.flink.table.api.DataTypes.GEOGRAPHY; import static org.apache.flink.table.api.DataTypes.INT; import static org.apache.flink.table.api.DataTypes.MAP; import static org.apache.flink.table.api.DataTypes.ROW; @@ -260,6 +261,62 @@ Stream getTestSetSpecs() { .testSqlValidationError( "CAST(CreateMultiset(f0) AS BITMAP)", "Cast function cannot convert value of type VARCHAR(2147483647) MULTISET to type BITMAP"), + TestSetSpec.forFunction(BuiltInFunctionDefinitions.CAST, "cast STRING to GEOGRAPHY") + .onFieldsWithData("POINT (0 0)") + .andDataTypes(STRING()) + .testTableApiValidationError( + $("f0").cast(GEOGRAPHY()), + "Unsupported cast from 'STRING' to 'GEOGRAPHY'") + .testSqlValidationError( + "CAST(f0 AS GEOGRAPHY)", + "Cast function cannot convert value of type VARCHAR(2147483647) to type GEOGRAPHY"), + TestSetSpec.forFunction(BuiltInFunctionDefinitions.CAST, "cast BYTES to GEOGRAPHY") + .onFieldsWithData(new byte[] {1, 2, 3}) + .andDataTypes(BYTES()) + .testTableApiValidationError( + $("f0").cast(GEOGRAPHY()), + "Unsupported cast from 'BYTES' to 'GEOGRAPHY'") + .testSqlValidationError( + "CAST(f0 AS GEOGRAPHY)", + "Cast function cannot convert value of type VARBINARY(2147483647) to type GEOGRAPHY"), + TestSetSpec.forFunction( + BuiltInFunctionDefinitions.CAST, "cast VARBINARY to GEOGRAPHY") + .onFieldsWithData(new byte[] {1, 2, 3}) + .andDataTypes(VARBINARY(3)) + .testTableApiValidationError( + $("f0").cast(GEOGRAPHY()), + "Unsupported cast from 'VARBINARY(3)' to 'GEOGRAPHY'") + .testSqlValidationError( + "CAST(f0 AS GEOGRAPHY)", + "Cast function cannot convert value of type VARBINARY(3) to type GEOGRAPHY"), + TestSetSpec.forFunction(BuiltInFunctionDefinitions.CAST, "cast GEOGRAPHY to STRING") + .onFieldsWithData((Object) null) + .andDataTypes(GEOGRAPHY()) + .testTableApiValidationError( + $("f0").cast(STRING()), + "Unsupported cast from 'GEOGRAPHY' to 'STRING'") + .testSqlValidationError( + "CAST(f0 AS STRING)", + "Cast function cannot convert value of type GEOGRAPHY to type VARCHAR(2147483647)"), + TestSetSpec.forFunction(BuiltInFunctionDefinitions.CAST, "cast GEOGRAPHY to BYTES") + .onFieldsWithData((Object) null) + .andDataTypes(GEOGRAPHY()) + .testTableApiValidationError( + $("f0").cast(BYTES()), + "Unsupported cast from 'GEOGRAPHY' to 'BYTES'") + .testSqlValidationError( + "CAST(f0 AS BYTES)", + "Cast function cannot convert value of type GEOGRAPHY to type VARBINARY(2147483647)"), + TestSetSpec.forFunction( + BuiltInFunctionDefinitions.CAST, "cast GEOGRAPHY to VARBINARY") + .onFieldsWithData((Object) null) + .andDataTypes(GEOGRAPHY()) + .testTableApiValidationError( + $("f0").cast(VARBINARY(3)), + "Unsupported cast from 'GEOGRAPHY' to 'VARBINARY(3)'") + .testSqlValidationError( + "CAST(f0 AS VARBINARY(3))", + "Cast function cannot convert value of type GEOGRAPHY to type VARBINARY(3)"), TestSetSpec.forFunction(BuiltInFunctionDefinitions.CAST, "cast RAW to STRING") .onFieldsWithData("2020-11-11T18:08:01.123") .andDataTypes(STRING()) diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DataTypeJsonSerdeTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DataTypeJsonSerdeTest.java index 3c24b472bac932..766edc9786837e 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DataTypeJsonSerdeTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DataTypeJsonSerdeTest.java @@ -59,6 +59,7 @@ private static Stream testDataTypeSerde() { DataTypes.TIMESTAMP_LTZ(3).toInternal(), DataTypes.TIMESTAMP_LTZ(9).bridgedTo(long.class), DataTypes.BITMAP(), + DataTypes.GEOGRAPHY(), DataTypes.ROW( DataTypes.TIMESTAMP_LTZ(3).toInternal(), DataTypes.TIMESTAMP_LTZ(9).bridgedTo(long.class), diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerdeTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerdeTest.java index 0742134ebd64bd..4e891871b9c0b8 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerdeTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerdeTest.java @@ -43,6 +43,7 @@ import org.apache.flink.table.types.logical.DescriptorType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.LogicalType; @@ -267,6 +268,7 @@ private static List testLogicalTypeSerde() { new MultisetType(BinaryType.ofEmptyLiteral()), new MultisetType(VarBinaryType.ofEmptyLiteral()), new BitmapType(), + new GeographyType(), RowType.of(new BigIntType(), new IntType(false), new VarCharType(200)), RowType.of( new LogicalType[] { diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/BoxedWrapperRowData.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/BoxedWrapperRowData.java index d5e7f361d7fa98..33030f2e35775c 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/BoxedWrapperRowData.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/BoxedWrapperRowData.java @@ -154,6 +154,11 @@ public Bitmap getBitmap(int pos) { return (Bitmap) this.fields[pos]; } + @Override + public GeographyData getGeography(int pos) { + return (GeographyData) this.fields[pos]; + } + @Override public void setNullAt(int pos) { this.fields[pos] = null; diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/RowDataTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/RowDataTest.java index 787e54957c7490..37e5d492a55774 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/RowDataTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/RowDataTest.java @@ -33,11 +33,13 @@ import org.apache.flink.table.types.logical.ArrayType; import org.apache.flink.table.types.logical.BigIntType; import org.apache.flink.table.types.logical.BinaryType; +import org.apache.flink.table.types.logical.BitmapType; import org.apache.flink.table.types.logical.BooleanType; import org.apache.flink.table.types.logical.CharType; import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.MapType; import org.apache.flink.table.types.logical.MultisetType; @@ -63,7 +65,11 @@ /** Test for {@link RowData}s. */ class RowDataTest { - private static final int NUM_FIELDS = 19; + private static final int NUM_FIELDS = 20; + private static final byte[] POINT_WKB = + new byte[] { + 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, (byte) 0xF0, 0x3F, 0, 0, 0, 0, 0, 0, 0, 0x40 + }; private StringData str; private RawValueData generic; @@ -77,6 +83,7 @@ class RowDataTest { private TimestampData timestamp1; private TimestampData timestamp2; private Bitmap bitmap; + private GeographyData geography; @BeforeEach void before() { @@ -105,6 +112,7 @@ void before() { timestamp2 = TimestampData.fromLocalDateTime(LocalDateTime.of(1969, 1, 1, 0, 0, 0, 123456789)); bitmap = Bitmap.fromArray(new int[] {1, 2, 3}); + geography = GeographyData.fromBytes(POINT_WKB); } @Test @@ -152,6 +160,7 @@ private BinaryRowData getBinaryRow() { writer.writeTimestamp(16, timestamp1, 3); writer.writeTimestamp(17, timestamp2, 9); writer.writeBitmap(18, bitmap); + writer.writeGeography(19, geography); return row; } @@ -177,6 +186,7 @@ void testGenericRow() { row.setField(16, timestamp1); row.setField(17, timestamp2); row.setField(18, bitmap); + row.setField(19, geography); testGetters(row); } @@ -201,6 +211,7 @@ public void testBoxedWrapperRow() { row.setNonPrimitiveValue(16, timestamp1); row.setNonPrimitiveValue(17, timestamp2); row.setNonPrimitiveValue(18, bitmap); + row.setNonPrimitiveValue(19, geography); testGetters(row); testSetters(row); } @@ -230,6 +241,7 @@ public void testJoinedRow() { row2.setField(11, timestamp1); row2.setField(12, timestamp2); row2.setField(13, bitmap); + row2.setField(14, geography); testGetters(new JoinedRowData(row1, row2)); } @@ -277,6 +289,14 @@ void testFieldGetters() { .isEqualTo(timestamp1); assertThat(RowData.createFieldGetter(new TimestampType(9), 17).getFieldOrNull(row)) .isEqualTo(timestamp2); + assertThat(RowData.createFieldGetter(new BitmapType(), 18).getFieldOrNull(row)) + .isEqualTo(bitmap); + assertThat( + ((GeographyData) + RowData.createFieldGetter(new GeographyType(), 19) + .getFieldOrNull(row)) + .toBytes()) + .isEqualTo(geography.toBytes()); } @Test @@ -355,6 +375,10 @@ private void testFieldGettersWithNull(boolean nullable) { RowData.createFieldGetter(new TimestampType(nullable, 9), 17) .getFieldOrNull(row)) .isNull(); + assertThat(RowData.createFieldGetter(new BitmapType(nullable), 18).getFieldOrNull(row)) + .isNull(); + assertThat(RowData.createFieldGetter(new GeographyType(nullable), 19).getFieldOrNull(row)) + .isNull(); } private void testGetters(RowData row) { @@ -385,6 +409,7 @@ private void testGetters(RowData row) { assertThat(row.getTimestamp(16, 3)).isEqualTo(timestamp1); assertThat(row.getTimestamp(17, 9)).isEqualTo(timestamp2); assertThat(row.getBitmap(18)).isEqualTo(bitmap); + assertThat(row.getGeography(19).toBytes()).isEqualTo(geography.toBytes()); } private void testSetters(RowData row) { @@ -435,7 +460,7 @@ private void testSetters(RowData row) { } private static BinaryRowData getNullBinaryRow() { - BinaryRowData row = new BinaryRowData(18); + BinaryRowData row = new BinaryRowData(NUM_FIELDS); BinaryRowWriter binaryRowWriter = new BinaryRowWriter(row); for (int i = 0; i < row.getArity(); i++) { binaryRowWriter.setNullAt(i); diff --git a/flink-table/flink-table-test-utils/pom.xml b/flink-table/flink-table-test-utils/pom.xml index 590f4d52359bff..9460a52e034aeb 100644 --- a/flink-table/flink-table-test-utils/pom.xml +++ b/flink-table/flink-table-test-utils/pom.xml @@ -59,6 +59,12 @@ under the License. flink-table-type-utils ${project.version} + + org.apache.flink + flink-table-type-utils + ${project.version} + test + org.apache.flink flink-table-planner-loader diff --git a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/DataStructureConverters.java b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/DataStructureConverters.java index f01a1aca779da7..3123edcb3b8442 100644 --- a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/DataStructureConverters.java +++ b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/DataStructureConverters.java @@ -22,11 +22,13 @@ import org.apache.flink.table.api.TableException; import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GeographyData; import org.apache.flink.table.data.MapData; import org.apache.flink.table.data.RawValueData; import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.StringData; import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.data.binary.BinaryGeographyData; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.LogicalTypeRoot; @@ -200,6 +202,8 @@ public final class DataStructureConverters { putConverter(LogicalTypeRoot.VARIANT, Variant.class, identity()); putConverter(LogicalTypeRoot.BITMAP, Bitmap.class, constructor(BitmapBitmapConverter::new)); putConverter(LogicalTypeRoot.BITMAP, RoaringBitmapData.class, identity()); + putConverter(LogicalTypeRoot.GEOGRAPHY, GeographyData.class, identity()); + putConverter(LogicalTypeRoot.GEOGRAPHY, BinaryGeographyData.class, identity()); } /** Returns a converter for the given {@link DataType}. */ diff --git a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/AbstractBinaryWriter.java b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/AbstractBinaryWriter.java index 73f817e64815eb..0564d0b0cc634e 100644 --- a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/AbstractBinaryWriter.java +++ b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/AbstractBinaryWriter.java @@ -24,6 +24,7 @@ import org.apache.flink.core.memory.MemorySegmentFactory; import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GeographyData; import org.apache.flink.table.data.MapData; import org.apache.flink.table.data.RawValueData; import org.apache.flink.table.data.RowData; @@ -141,6 +142,11 @@ public void writeBitmap(int pos, Bitmap bitmap) { writeBytesToVarLenPart(pos, bytes, bytes.length); } + @Override + public void writeGeography(int pos, GeographyData geography) { + writeBinary(pos, geography.toBytes()); + } + private DataOutputViewStreamWrapper getOutputView() { if (outputView == null) { outputView = new DataOutputViewStreamWrapper(new BinaryRowWriterOutputView()); diff --git a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryArrayWriter.java b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryArrayWriter.java index 857c2ecd42466f..68540b9fc93e25 100644 --- a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryArrayWriter.java +++ b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryArrayWriter.java @@ -247,6 +247,7 @@ public static NullSetter createNullSetter(LogicalType elementType) { case RAW: case VARIANT: case BITMAP: + case GEOGRAPHY: return BinaryArrayWriter::setNullLong; case BOOLEAN: return BinaryArrayWriter::setNullBoolean; diff --git a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryWriter.java b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryWriter.java index 4db8977ccd2f0b..482540b1d53312 100644 --- a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryWriter.java +++ b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryWriter.java @@ -21,6 +21,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GeographyData; import org.apache.flink.table.data.MapData; import org.apache.flink.table.data.RawValueData; import org.apache.flink.table.data.RowData; @@ -91,6 +92,8 @@ public interface BinaryWriter { void writeBitmap(int pos, Bitmap bitmap); + void writeGeography(int pos, GeographyData geography); + /** Finally, complete write to set real size to binary. */ void complete(); @@ -174,6 +177,9 @@ static void write( case BITMAP: writer.writeBitmap(pos, (Bitmap) o); break; + case GEOGRAPHY: + writer.writeGeography(pos, (GeographyData) o); + break; default: throw new UnsupportedOperationException("Not support type: " + type); } @@ -253,6 +259,8 @@ static ValueSetter createValueSetter(LogicalType elementType) { return (writer, pos, value) -> writer.writeVariant(pos, (Variant) value); case BITMAP: return (writer, pos, value) -> writer.writeBitmap(pos, (Bitmap) value); + case GEOGRAPHY: + return (writer, pos, value) -> writer.writeGeography(pos, (GeographyData) value); case NULL: case SYMBOL: case UNRESOLVED: diff --git a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/GeographyTypeSerializer.java b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/GeographyTypeSerializer.java new file mode 100644 index 00000000000000..3e54f24009eb82 --- /dev/null +++ b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/GeographyTypeSerializer.java @@ -0,0 +1,127 @@ +/* + * 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.flink.table.runtime.typeutils; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton; +import org.apache.flink.core.memory.DataInputView; +import org.apache.flink.core.memory.DataOutputView; +import org.apache.flink.table.data.GeographyData; + +import java.io.IOException; + +/** + * Serializer for {@link GeographyData} values of {@link + * org.apache.flink.table.types.logical.GeographyType}. + */ +@Internal +public final class GeographyTypeSerializer extends TypeSerializerSingleton { + + private static final long serialVersionUID = 1L; + + private static final int FORMAT_VERSION = 1; + + private static final byte[] EMPTY_GEOMETRY_COLLECTION = + new byte[] {1, GeographyData.GEOMETRY_COLLECTION, 0, 0, 0, 0, 0, 0, 0}; + + public static final GeographyTypeSerializer INSTANCE = new GeographyTypeSerializer(); + + private GeographyTypeSerializer() {} + + @Override + public boolean isImmutableType() { + return true; + } + + @Override + public GeographyData createInstance() { + return GeographyData.fromBytes(EMPTY_GEOMETRY_COLLECTION); + } + + @Override + public GeographyData copy(GeographyData from) { + return GeographyData.fromBytes(from.toBytes()); + } + + @Override + public GeographyData copy(GeographyData from, GeographyData reuse) { + return copy(from); + } + + @Override + public int getLength() { + return -1; + } + + @Override + public void serialize(GeographyData record, DataOutputView target) throws IOException { + final byte[] bytes = record.toBytes(); + target.writeByte(FORMAT_VERSION); + target.writeInt(bytes.length); + target.write(bytes); + } + + @Override + public GeographyData deserialize(DataInputView source) throws IOException { + readFormatVersion(source); + final int length = readPayloadLength(source); + final byte[] bytes = new byte[length]; + source.readFully(bytes); + return GeographyData.fromBytes(bytes); + } + + @Override + public GeographyData deserialize(GeographyData reuse, DataInputView source) throws IOException { + return deserialize(source); + } + + @Override + public void copy(DataInputView source, DataOutputView target) throws IOException { + final int version = readFormatVersion(source); + final int length = readPayloadLength(source); + target.writeByte(version); + target.writeInt(length); + target.write(source, length); + } + + @Override + public TypeSerializerSnapshot snapshotConfiguration() { + return new GeographyTypeSerializerSnapshot(); + } + + private static int readFormatVersion(DataInputView source) throws IOException { + final int version = source.readUnsignedByte(); + if (version != FORMAT_VERSION) { + throw new IOException( + String.format( + "Unsupported GEOGRAPHY serializer format version %d. Expected %d.", + version, FORMAT_VERSION)); + } + return version; + } + + private static int readPayloadLength(DataInputView source) throws IOException { + final int length = source.readInt(); + if (length < 0) { + throw new IOException(String.format("Invalid GEOGRAPHY payload length %d.", length)); + } + return length; + } +} diff --git a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/GeographyTypeSerializerSnapshot.java b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/GeographyTypeSerializerSnapshot.java new file mode 100644 index 00000000000000..a3159016479899 --- /dev/null +++ b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/GeographyTypeSerializerSnapshot.java @@ -0,0 +1,63 @@ +/* + * 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.flink.table.runtime.typeutils; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.core.memory.DataInputView; +import org.apache.flink.core.memory.DataOutputView; +import org.apache.flink.table.data.GeographyData; + +import java.io.IOException; + +/** Serializer snapshot for {@link GeographyTypeSerializer}. */ +@Internal +public final class GeographyTypeSerializerSnapshot + implements TypeSerializerSnapshot { + + private static final int CURRENT_VERSION = 1; + + @Override + public int getCurrentVersion() { + return CURRENT_VERSION; + } + + @Override + public void writeSnapshot(DataOutputView out) throws IOException {} + + @Override + public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) + throws IOException {} + + @Override + public TypeSerializer restoreSerializer() { + return GeographyTypeSerializer.INSTANCE; + } + + @Override + public TypeSerializerSchemaCompatibility resolveSchemaCompatibility( + TypeSerializerSnapshot oldSerializerSnapshot) { + if (oldSerializerSnapshot instanceof GeographyTypeSerializerSnapshot) { + return TypeSerializerSchemaCompatibility.compatibleAsIs(); + } + return TypeSerializerSchemaCompatibility.incompatible(); + } +} diff --git a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalSerializers.java b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalSerializers.java index 371dcd136e1d56..dc99d6477d7d5c 100644 --- a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalSerializers.java +++ b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalSerializers.java @@ -129,6 +129,8 @@ private static TypeSerializer createInternal(LogicalType type) { return VariantSerializer.INSTANCE; case BITMAP: return BitmapSerializer.INSTANCE; + case GEOGRAPHY: + return GeographyTypeSerializer.INSTANCE; case NULL: case SYMBOL: case UNRESOLVED: diff --git a/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/GeographyTypeSerializerTest.java b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/GeographyTypeSerializerTest.java new file mode 100644 index 00000000000000..8a695c5c020f54 --- /dev/null +++ b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/GeographyTypeSerializerTest.java @@ -0,0 +1,215 @@ +/* + * 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.flink.table.runtime.typeutils; + +import org.apache.flink.api.common.typeutils.SerializerTestBase; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil; +import org.apache.flink.core.memory.DataInputViewStreamWrapper; +import org.apache.flink.core.memory.DataOutputViewStreamWrapper; +import org.apache.flink.table.data.GenericArrayData; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.GeographyData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.binary.BinaryArrayData; +import org.apache.flink.table.data.binary.BinaryRowData; +import org.apache.flink.table.types.logical.GeographyType; + +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link GeographyTypeSerializer}. */ +class GeographyTypeSerializerTest extends SerializerTestBase { + + private static final int FORMAT_VERSION = 1; + + private static final byte[] POINT_WKB = + new byte[] { + 1, GeographyData.POINT, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 + }; + + private static final byte[] BIG_ENDIAN_POINT_WKB = + new byte[] { + 0, 0, 0, 0, GeographyData.POINT, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 + }; + + @Override + protected TypeSerializer createSerializer() { + return GeographyTypeSerializer.INSTANCE; + } + + @Override + protected int getLength() { + return -1; + } + + @Override + protected Class getTypeClass() { + return GeographyData.class; + } + + @Override + protected GeographyData[] getTestData() { + return new GeographyData[] { + GeographyData.fromBytes(POINT_WKB), + GeographyData.fromBytes(BIG_ENDIAN_POINT_WKB), + GeographyTypeSerializer.INSTANCE.createInstance() + }; + } + + @Override + protected void deepEquals(String message, GeographyData should, GeographyData is) { + assertThat(is.toBytes()).as(message).isEqualTo(should.toBytes()); + } + + @Test + void testInternalSerializerRoundTripsRawWkb() throws Exception { + final TypeSerializer serializer = + InternalSerializers.create(new GeographyType()); + final GeographyData geography = GeographyData.fromBytes(POINT_WKB); + final ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + + serializer.serialize(geography, new DataOutputViewStreamWrapper(bytes)); + final GeographyData deserialized = + serializer.deserialize( + new DataInputViewStreamWrapper( + new ByteArrayInputStream(bytes.toByteArray()))); + + assertThat(deserialized.toBytes()).isEqualTo(POINT_WKB); + assertThat(deserialized.subtypeId()).isEqualTo(GeographyData.POINT); + } + + @Test + void testSerializedFormUsesVersionedLengthEnvelope() throws Exception { + final ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + + GeographyTypeSerializer.INSTANCE.serialize( + GeographyData.fromBytes(POINT_WKB), new DataOutputViewStreamWrapper(bytes)); + final DataInputViewStreamWrapper input = + new DataInputViewStreamWrapper(new ByteArrayInputStream(bytes.toByteArray())); + final byte[] payload = new byte[POINT_WKB.length]; + + assertThat(input.readUnsignedByte()).isEqualTo(FORMAT_VERSION); + assertThat(input.readInt()).isEqualTo(POINT_WKB.length); + input.readFully(payload); + assertThat(payload).isEqualTo(POINT_WKB); + } + + @Test + void testRejectsUnsupportedPayloadVersion() throws Exception { + final ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + final DataOutputViewStreamWrapper output = new DataOutputViewStreamWrapper(bytes); + output.writeByte(2); + output.writeInt(POINT_WKB.length); + output.write(POINT_WKB); + + assertThatThrownBy( + () -> + GeographyTypeSerializer.INSTANCE.deserialize( + new DataInputViewStreamWrapper( + new ByteArrayInputStream(bytes.toByteArray())))) + .isInstanceOf(IOException.class) + .hasMessageContaining("Unsupported GEOGRAPHY serializer format version 2"); + } + + @Test + void testSnapshotSelfCompatibility() throws Exception { + final TypeSerializerSnapshot snapshot = + GeographyTypeSerializer.INSTANCE.snapshotConfiguration(); + final ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + + TypeSerializerSnapshotSerializationUtil.writeSerializerSnapshot( + new DataOutputViewStreamWrapper(bytes), snapshot); + final TypeSerializerSnapshot restoredSnapshot = + TypeSerializerSnapshotSerializationUtil.readSerializerSnapshot( + new DataInputViewStreamWrapper( + new ByteArrayInputStream(bytes.toByteArray())), + getClass().getClassLoader()); + + assertThat( + GeographyTypeSerializer.INSTANCE + .snapshotConfiguration() + .resolveSchemaCompatibility(restoredSnapshot) + .isCompatibleAsIs()) + .isTrue(); + assertThat(restoredSnapshot.restoreSerializer()).isSameAs(GeographyTypeSerializer.INSTANCE); + } + + @Test + void testCopyPreservesRawWkbBytes() { + final GeographyData geography = GeographyData.fromBytes(POINT_WKB); + final GeographyData copied = GeographyTypeSerializer.INSTANCE.copy(geography); + + assertThat(copied).isNotSameAs(geography); + assertThat(copied.toBytes()).isEqualTo(POINT_WKB); + assertThat(copied.subtypeId()).isEqualTo(GeographyData.POINT); + } + + @Test + void testCreateInstanceReturnsValidGeographyData() { + final GeographyData instance = GeographyTypeSerializer.INSTANCE.createInstance(); + + assertThat(instance.subtypeId()).isEqualTo(GeographyData.GEOMETRY_COLLECTION); + assertThat(instance.sizeInBytes()).isEqualTo(9); + } + + @Test + void testRowDataSerializerConvertsGenericRowToBinaryRow() { + final RowDataSerializer serializer = + InternalTypeInfo.ofFields(new GeographyType()).toRowSerializer(); + final GeographyData geography = GeographyData.fromBytes(POINT_WKB); + final GenericRowData row = GenericRowData.of(geography); + + final BinaryRowData binaryRow = serializer.toBinaryRow(row, true); + + assertThat(binaryRow.getGeography(0).toBytes()).isEqualTo(POINT_WKB); + assertThat(binaryRow.getGeography(0).subtypeId()).isEqualTo(GeographyData.POINT); + } + + @Test + void testRowDataSerializerPreservesNullGeography() { + final RowDataSerializer serializer = + InternalTypeInfo.ofFields(new GeographyType()).toRowSerializer(); + final GenericRowData row = GenericRowData.of((Object) null); + + final BinaryRowData binaryRow = serializer.toBinaryRow(row, true); + + assertThat(binaryRow.isNullAt(0)).isTrue(); + } + + @Test + void testArrayDataSerializerConvertsGenericArrayToBinaryArray() { + final ArrayDataSerializer serializer = new ArrayDataSerializer(new GeographyType()); + final GenericArrayData array = + new GenericArrayData(new Object[] {GeographyData.fromBytes(POINT_WKB), null}); + + final BinaryArrayData binaryArray = serializer.toBinaryArray(array); + + assertThat(binaryArray.getGeography(0).toBytes()).isEqualTo(POINT_WKB); + assertThat(binaryArray.getGeography(0).subtypeId()).isEqualTo(GeographyData.POINT); + assertThat(binaryArray.isNullAt(1)).isTrue(); + } +} diff --git a/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataSerializerTest.java b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataSerializerTest.java index 6d6379ed046965..e5a09b407e45b6 100644 --- a/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataSerializerTest.java +++ b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataSerializerTest.java @@ -24,6 +24,7 @@ import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.GeographyData; import org.apache.flink.table.data.RawValueData; import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.StringData; @@ -41,6 +42,7 @@ import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.GeographyType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.LogicalType; @@ -62,6 +64,11 @@ /** Test for {@link RowDataSerializer}. */ abstract class RowDataSerializerTest extends SerializerTestInstance { + private static final byte[] POINT_WKB = + new byte[] { + 1, GeographyData.POINT, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 + }; + private final RowDataSerializer serializer; private final RowData[] testData; @@ -233,6 +240,28 @@ private static RowDataSerializer getRowSerializer() { } } + static final class RowDataSerializerWithGeographyTest extends RowDataSerializerTest { + public RowDataSerializerWithGeographyTest() { + super(getRowSerializer(), getData()); + } + + private static RowData[] getData() { + GenericRowData row1 = new GenericRowData(1); + row1.setField(0, GeographyData.fromBytes(POINT_WKB)); + + GenericRowData row2 = new GenericRowData(1); + row2.setField(0, null); + + return new RowData[] {row1, row2}; + } + + private static RowDataSerializer getRowSerializer() { + InternalTypeInfo typeInfo = InternalTypeInfo.ofFields(new GeographyType()); + + return typeInfo.toRowSerializer(); + } + } + static final class LargeRowDataSerializerTest extends RowDataSerializerTest { public LargeRowDataSerializerTest() { super(getRowSerializer(), getData()); diff --git a/flink-tests/pom.xml b/flink-tests/pom.xml index 4ef3c5f273d53c..ee49aca935c6d1 100644 --- a/flink-tests/pom.xml +++ b/flink-tests/pom.xml @@ -69,6 +69,13 @@ under the License. test + + org.apache.flink + flink-connector-base + ${project.version} + test + + org.apache.flink flink-connector-files @@ -205,6 +212,13 @@ under the License. test + + org.apache.flink + flink-table-type-utils + ${project.version} + test + + org.apache.flink flink-runtime diff --git a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java index 0ac026145139e6..84ada2f5b4271e 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java +++ b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java @@ -83,6 +83,7 @@ import org.apache.flink.table.runtime.typeutils.BinaryRowDataSerializer; import org.apache.flink.table.runtime.typeutils.DecimalDataSerializer; import org.apache.flink.table.runtime.typeutils.ExternalSerializer; +import org.apache.flink.table.runtime.typeutils.GeographyTypeSerializer; import org.apache.flink.table.runtime.typeutils.LinkedListSerializer; import org.apache.flink.table.runtime.typeutils.MapDataSerializer; import org.apache.flink.table.runtime.typeutils.RawValueDataSerializer; @@ -196,6 +197,7 @@ void testTypeSerializerTestCoverage() { SharedBufferEdge.SharedBufferEdgeSerializer.class.getName(), RowDataSerializer.class.getName(), DecimalDataSerializer.class.getName(), + GeographyTypeSerializer.class.getName(), AvroSerializer.class.getName()); // type serializer whitelist for TypeSerializerUpgradeTestBase test coverage @@ -258,6 +260,7 @@ void testTypeSerializerTestCoverage() { SharedBufferEdge.SharedBufferEdgeSerializer.class.getName(), RowDataSerializer.class.getName(), DecimalDataSerializer.class.getName(), + GeographyTypeSerializer.class.getName(), AvroSerializer.class.getName(), // KeyAndValueSerializer shouldn't be used to serialize data to state and // doesn't need to ensure upgrade compatibility.