Skip to content

Commit d366b1e

Browse files
exAspArkexAspArk
committed
WIP
1 parent 989c88e commit d366b1e

8 files changed

Lines changed: 85 additions & 69 deletions

File tree

‎src/duckdb.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ import (
1010
"strings"
1111
"time"
1212

13-
_ "github.com/marcboeker/go-duckdb"
13+
_ "github.com/marcboeker/go-duckdb/v2"
1414
)
1515

1616
const (
@@ -94,7 +94,9 @@ func NewDuckdb(config *Config, withPgCompatibility bool) *Duckdb {
9494
}
9595

9696
if config.LogLevel == LOG_LEVEL_TRACE {
97-
_, err = duckdb.ExecContext(ctx, "SET enable_http_logging=true", nil)
97+
_, err = duckdb.ExecContext(ctx, "PRAGMA enable_logging('HTTP')", nil)
98+
PanicIfError(config, err)
99+
_, err = duckdb.ExecContext(ctx, "SET logging_storage = 'stdout'", nil)
98100
PanicIfError(config, err)
99101
}
100102
}

‎src/go.mod‎

Lines changed: 20 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -11,20 +11,20 @@ require (
1111
github.com/google/uuid v1.6.0
1212
github.com/jackc/pgx/v5 v5.7.2
1313
github.com/linkedin/goavro v2.1.0+incompatible
14-
github.com/marcboeker/go-duckdb v1.8.3
1514
github.com/pganalyze/pg_query_go/v5 v5.1.0
1615
github.com/xitongsys/parquet-go v1.6.3-0.20240813051905-693d3323dee0
1716
)
1817

1918
require (
19+
github.com/marcboeker/go-duckdb/v2 v2.3.2
2020
github.com/xitongsys/parquet-go-source v0.0.0-20241021075129-b732d2ac9c9b
2121
golang.org/x/crypto v0.35.0
22-
golang.org/x/exp v0.0.0-20240909161429-701f63a606c0
22+
golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c
2323
)
2424

2525
require (
2626
github.com/andybalholm/brotli v1.1.1 // indirect
27-
github.com/apache/arrow-go/v18 v18.0.0 // indirect
27+
github.com/apache/arrow-go/v18 v18.1.0 // indirect
2828
github.com/apache/arrow/go/v12 v12.0.1 // indirect
2929
github.com/apache/thrift v0.21.0 // indirect
3030
github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.6 // indirect
@@ -41,26 +41,34 @@ require (
4141
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.3 // indirect
4242
github.com/aws/aws-sdk-go-v2/service/sts v1.32.3 // indirect
4343
github.com/aws/smithy-go v1.22.0 // indirect
44-
github.com/goccy/go-json v0.10.3 // indirect
44+
github.com/duckdb/duckdb-go-bindings v0.1.16 // indirect
45+
github.com/duckdb/duckdb-go-bindings/darwin-amd64 v0.1.11 // indirect
46+
github.com/duckdb/duckdb-go-bindings/darwin-arm64 v0.1.11 // indirect
47+
github.com/duckdb/duckdb-go-bindings/linux-amd64 v0.1.11 // indirect
48+
github.com/duckdb/duckdb-go-bindings/linux-arm64 v0.1.11 // indirect
49+
github.com/duckdb/duckdb-go-bindings/windows-amd64 v0.1.11 // indirect
50+
github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
51+
github.com/goccy/go-json v0.10.5 // indirect
4552
github.com/goccy/go-reflect v1.2.0 // indirect
4653
github.com/golang/snappy v0.0.4 // indirect
47-
github.com/google/flatbuffers v24.3.25+incompatible // indirect
54+
github.com/google/flatbuffers v25.1.24+incompatible // indirect
4855
github.com/jackc/pgpassfile v1.0.0 // indirect
4956
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
5057
github.com/klauspost/asmfmt v1.3.2 // indirect
5158
github.com/klauspost/compress v1.17.11 // indirect
52-
github.com/klauspost/cpuid/v2 v2.2.8 // indirect
59+
github.com/klauspost/cpuid/v2 v2.2.9 // indirect
60+
github.com/marcboeker/go-duckdb/arrowmapping v0.0.9 // indirect
61+
github.com/marcboeker/go-duckdb/mapping v0.0.10 // indirect
5362
github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 // indirect
5463
github.com/minio/c2goasm v0.0.0-20190812172519-36a3d3bbc4f3 // indirect
55-
github.com/mitchellh/mapstructure v1.5.0 // indirect
56-
github.com/pierrec/lz4/v4 v4.1.21 // indirect
64+
github.com/pierrec/lz4/v4 v4.1.22 // indirect
5765
github.com/zeebo/xxh3 v1.0.2 // indirect
58-
golang.org/x/mod v0.21.0 // indirect
66+
golang.org/x/mod v0.22.0 // indirect
5967
golang.org/x/sync v0.11.0 // indirect
6068
golang.org/x/sys v0.30.0 // indirect
6169
golang.org/x/text v0.22.0 // indirect
62-
golang.org/x/tools v0.26.0 // indirect
63-
golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
64-
google.golang.org/protobuf v1.35.1 // indirect
70+
golang.org/x/tools v0.29.0 // indirect
71+
golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
72+
google.golang.org/protobuf v1.36.1 // indirect
6573
gopkg.in/linkedin/goavro.v1 v1.0.5 // indirect
6674
)

‎src/go.sum‎

Lines changed: 42 additions & 26 deletions
Large diffs are not rendered by default.

‎src/iceberg_writer_table_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1394,7 +1394,7 @@ func testRecords(t *testing.T, duckdb *Duckdb, expectedRecords [][]string) {
13941394
icebergReader := NewIcebergReader(duckdb.config)
13951395
metadataFilePath := icebergReader.MetadataFilePath(TEST_ICEBERG_WRITER_SCHEMA_TABLE)
13961396

1397-
rows, err := duckdb.QueryContext(context.Background(), "SELECT id::text, COALESCE(name, '"+PG_NULL_STRING+"') FROM iceberg_scan('"+metadataFilePath+"', skip_schema_inference = true) ORDER BY id")
1397+
rows, err := duckdb.QueryContext(context.Background(), "SELECT id::text, COALESCE(name, '"+PG_NULL_STRING+"') FROM iceberg_scan('"+metadataFilePath+"') ORDER BY id")
13981398
if err != nil {
13991399
t.Fatalf("Error querying DuckDB: %v", err)
14001400
}

‎src/parser_table.go‎

Lines changed: 2 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -42,8 +42,8 @@ func (parser *ParserTable) IsTableFromInformationSchema(qSchemaTable QuerySchema
4242
return qSchemaTable.Schema == PG_SCHEMA_INFORMATION_SCHEMA
4343
}
4444

45-
// public.table -> FROM iceberg_scan('path', skip_schema_inference = true) table
46-
// schema.table -> FROM iceberg_scan('path', skip_schema_inference = true) schema_table
45+
// public.table -> FROM iceberg_scan('path') table
46+
// schema.table -> FROM iceberg_scan('path') schema_table
4747
func (parser *ParserTable) MakeIcebergTableNode(queryToIcebergTable QueryToIcebergTable) *pgQuery.Node {
4848
node := pgQuery.MakeSimpleRangeFunctionNode([]*pgQuery.Node{
4949
pgQuery.MakeListNode([]*pgQuery.Node{
@@ -56,13 +56,6 @@ func (parser *ParserTable) MakeIcebergTableNode(queryToIcebergTable QueryToIcebe
5656
queryToIcebergTable.IcebergTablePath,
5757
0,
5858
),
59-
pgQuery.MakeAExprNode(
60-
pgQuery.A_Expr_Kind_AEXPR_OP,
61-
[]*pgQuery.Node{pgQuery.MakeStrNode("=")},
62-
pgQuery.MakeColumnRefNode([]*pgQuery.Node{pgQuery.MakeStrNode("skip_schema_inference")}, 0),
63-
parser.utils.MakeAConstBoolNode(true),
64-
0,
65-
),
6659
},
6760
0,
6861
),

‎src/query_handler.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ import (
1313
"github.com/google/uuid"
1414
"github.com/jackc/pgx/v5/pgproto3"
1515
"github.com/jackc/pgx/v5/pgtype"
16-
duckDb "github.com/marcboeker/go-duckdb"
16+
duckDb "github.com/marcboeker/go-duckdb/v2"
1717
)
1818

1919
const (

‎src/query_remapper_table.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -167,8 +167,8 @@ func (remapper *QueryRemapperTable) RemapTable(node *pgQuery.Node) *pgQuery.Node
167167
return node
168168
}
169169

170-
// public.table -> FROM iceberg_scan('path', skip_schema_inference = true) table
171-
// schema.table -> FROM iceberg_scan('path', skip_schema_inference = true) schema_table
170+
// public.table -> FROM iceberg_scan('path') table
171+
// schema.table -> FROM iceberg_scan('path') schema_table
172172
schemaTable := qSchemaTable.ToIcebergSchemaTable()
173173
if !remapper.icebergSchemaTables.Contains(schemaTable) { // Reload Iceberg tables if not found
174174
remapper.reloadIceberSchemaTables()

‎src/syncer.go‎

Lines changed: 13 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -111,9 +111,9 @@ func (syncer *Syncer) WriteInternalStartSqlFile(pgSchemaTables []PgSchemaTable)
111111

112112
for parent, children := range childTablesByParentTable {
113113
// CREATE OR REPLACE TABLE test_table AS
114-
// SELECT * FROM iceberg_scan('/iceberg/public/test_table_q1/metadata/v1.metadata.json', skip_schema_inference = true)
114+
// SELECT * FROM iceberg_scan('/iceberg/public/test_table_q1/metadata/v1.metadata.json')
115115
// UNION ALL
116-
// SELECT * FROM iceberg_scan('/iceberg/public/test_table_q2/metadata/v1.metadata.json', skip_schema_inference = true)
116+
// SELECT * FROM iceberg_scan('/iceberg/public/test_table_q2/metadata/v1.metadata.json')
117117

118118
subqueries := []string{}
119119
for _, child := range children {
@@ -129,36 +129,33 @@ func (syncer *Syncer) WriteInternalStartSqlFile(pgSchemaTables []PgSchemaTable)
129129
}
130130

131131
// Example:
132-
// - From postgres://username:pas$:wor^d@host:port/database
133-
// - To postgres://username:pas%24%3Awor%5Ed@host:port/database
134-
func (syncer *Syncer) urlEncodePassword(databaseUrl string) string {
132+
// - From postgres://username:pas$:wor^d#@host:port/database
133+
// - To postgres://username:pas%24%3Awor%5Ed%23@host:port/database
134+
func (syncer *Syncer) urlEncodePassword(pgDatabaseUrl string) string {
135135
// No credentials
136-
if !strings.Contains(databaseUrl, "@") {
137-
return databaseUrl
136+
if !strings.Contains(pgDatabaseUrl, "@") {
137+
return pgDatabaseUrl
138138
}
139139

140-
password := strings.TrimPrefix(databaseUrl, "postgresql://")
140+
password := strings.TrimPrefix(pgDatabaseUrl, "postgresql://")
141141
password = strings.TrimPrefix(password, "postgres://")
142142
passwordEndIndex := strings.LastIndex(password, "@")
143143
password = password[:passwordEndIndex]
144144

145145
// Credentials without password
146146
if !strings.Contains(password, ":") {
147-
return databaseUrl
147+
return pgDatabaseUrl
148148
}
149149

150150
_, password, _ = strings.Cut(password, ":")
151-
decodedPassword, err := url.QueryUnescape(password)
152-
if err != nil {
153-
return databaseUrl
154-
}
151+
encodedPassword := url.QueryEscape(password)
155152

156153
// Password is already encoded
157-
if decodedPassword != password {
158-
return databaseUrl
154+
if encodedPassword == password {
155+
return pgDatabaseUrl
159156
}
160157

161-
return strings.Replace(databaseUrl, ":"+password+"@", ":"+url.QueryEscape(password)+"@", 1)
158+
return strings.Replace(pgDatabaseUrl, ":"+password+"@", ":"+encodedPassword+"@", 1)
162159
}
163160

164161
func (syncer *Syncer) shouldSyncTable(pgSchemaTable PgSchemaTable) bool {

0 commit comments

Comments
 (0)