Avro array datatype support

Signed-off-by: Daniel Henneberger <git@danielhenneberger.com>
DataSQRL · Nov 8, 2024 · 71ecfde · 71ecfde
1 parent 0681be4
commit 71ecfde
Show file tree

Hide file tree

Showing 14 changed files with 82 additions and 486 deletions.
diff --git a/sqrl-calcite/src/main/java/com/datasqrl/calcite/dialect/ExtendedPostgresSqlDialect.java b/sqrl-calcite/src/main/java/com/datasqrl/calcite/dialect/ExtendedPostgresSqlDialect.java
@@ -99,7 +99,7 @@ public SqlDataTypeSpec getCastSpec(RelDataType type) {
           castSpec = "TIMESTAMP WITH TIME ZONE";
           break;
         case ARRAY:
-          castSpec = getCastSpec(type.getComponentType()) + "[]";
+          castSpec = "jsonb";
           break;
         case BINARY:
         case VARBINARY:

diff --git a/sqrl-flink-lib/sqrl-jdbc-1.19/src/main/java/com/datasqrl/jdbc/SqrlBaseJdbcRowConverter.java b/sqrl-flink-lib/sqrl-jdbc-1.19/src/main/java/com/datasqrl/jdbc/SqrlBaseJdbcRowConverter.java
@@ -50,10 +50,6 @@ protected JdbcSerializationConverter wrapIntoNullableExternalConverter(
           jdbcSerializationConverter.serialize(val, index, statement);
         }
       };
-    } else if (type.getTypeRoot() == ROW) {
-      return (val, index, statement) -> setRow(type, val, index, statement);
-    } else if (type.getTypeRoot() == MAP) {
-      return (val, index, statement) -> setRow(type, val, index, statement);
     }
     return super.wrapIntoNullableExternalConverter(jdbcSerializationConverter, type);
   }
@@ -87,30 +83,13 @@ protected JdbcSerializationConverter createExternalConverter(LogicalType type) {
         return (val, index, statement) ->
             statement.setTimestamp(
                 index, val.getTimestamp(index, tsPrecision).toTimestamp());
-      case ARRAY:
-        return (val, index, statement) -> setArray(type, val, index, statement);
-      case ROW:
-        return (val, index, statement) -> setRow(type, val, index, statement);
-      case MAP:
-        return (val, index, statement) -> setRow(type, val, index, statement);
       case MULTISET:
       case RAW:
       default:
         return super.createExternalConverter(type);
     }
   }
 
-  public abstract void setRow(LogicalType type, RowData val, int index,
-      FieldNamedPreparedStatement statement);
-  @SneakyThrows
-  public void setArray(LogicalType type, RowData val, int index, FieldNamedPreparedStatement statement) {
-    SqrlFieldNamedPreparedStatementImpl flinkPreparedStatement = (SqrlFieldNamedPreparedStatementImpl) statement;
-    for (int idx : flinkPreparedStatement.getIndexMapping()[index]) {
-      ArrayData arrayData = val.getArray(index);
-      createSqlArrayObject(type, arrayData, idx, flinkPreparedStatement.getStatement());
-    }
-  }
-
   @SneakyThrows
   private void createSqlArrayObject(LogicalType type, ArrayData data, int idx,
       PreparedStatement statement) {

diff --git a/...b/sqrl-jdbc-1.19/src/main/java/com/datasqrl/jdbc/SqrlFieldNamedPreparedStatementImpl.java b/...b/sqrl-jdbc-1.19/src/main/java/com/datasqrl/jdbc/SqrlFieldNamedPreparedStatementImpl.java
diff --git a/sqrl-flink-lib/sqrl-jdbc-1.19/src/main/java/com/datasqrl/jdbc/SqrlPostgresRowConverter.java b/sqrl-flink-lib/sqrl-jdbc-1.19/src/main/java/com/datasqrl/jdbc/SqrlPostgresRowConverter.java
@@ -63,19 +63,6 @@ public SqrlPostgresRowConverter(RowType rowType) {
         super(rowType);
     }
 
-    @SneakyThrows
-    public void setRow(LogicalType type, RowData val, int index,
-        FieldNamedPreparedStatement statement) {
-        SqrlFieldNamedPreparedStatementImpl flinkPreparedStatement = (SqrlFieldNamedPreparedStatementImpl) statement;
-        for (int idx : flinkPreparedStatement.getIndexMapping()[index]) {
-//            RowData row = val.getRow(index, ((RowType) type).getFieldCount());
-//            java.sql.Array sqlArray = flinkPreparedStatement.getStatement()
-//                .getConnection().createArrayOf("bytea", );
-            flinkPreparedStatement.getStatement().setBytes(idx, new byte[0]);
-        }
-    }
-
-
     @Override
     public JdbcDeserializationConverter createInternalConverter(LogicalType type) {
         if (sqrlSerializers.containsKey(type.getDefaultConversion())) {