Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Avro array datatype support #982

Merged
merged 8 commits into from
Nov 9, 2024
Merged

Avro array datatype support #982

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
93523f0
Support avro row type
henneberger Nov 8, 2024
888f34d
Fix snap
henneberger Nov 8, 2024
83c2d6c
Fix row walking
henneberger Nov 8, 2024
1011b57
Add avro single row datatype support
henneberger Nov 8, 2024
d5ffe79
Fix snaps
henneberger Nov 8, 2024
0681be4
Fix serialization for json
henneberger Nov 8, 2024
71ecfde
Avro array datatype support
henneberger Nov 8, 2024
3cd99aa
Fix snap
henneberger Nov 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion sqrl-calcite/src/main/java/com/datasqrl/calcite/dialect/ExtendedPostgresSqlDialect.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
21 changes: 0 additions & 21 deletions sqrl-flink-lib/sqrl-jdbc-1.19/src/main/java/com/datasqrl/jdbc/SqrlBaseJdbcRowConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down Expand Up @@ -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) {
Expand Down
252 changes: 0 additions & 252 deletions ...b/sqrl-jdbc-1.19/src/main/java/com/datasqrl/jdbc/SqrlFieldNamedPreparedStatementImpl.java
View file
Open in desktop

This file was deleted.

13 changes: 0 additions & 13 deletions sqrl-flink-lib/sqrl-jdbc-1.19/src/main/java/com/datasqrl/jdbc/SqrlPostgresRowConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())) {
Expand Down
Loading
Loading