Skip to content
This repository was archived by the owner on Dec 31, 2025. It is now read-only.

Commit e9e3579

Browse files
committed
chore: upgrade iceberg-go dependency and rename command append to add
1 parent a18d20c commit e9e3579

22 files changed

Lines changed: 424 additions & 169 deletions

File tree

Makefile

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,11 @@ bundle: build
1515
COPYFILE_DISABLE=1 tar --no-xattr -cvzf ${BUNDLE_PATH}/../bundle.tar.gz -C ${BUNDLE_PATH} .
1616

1717
test:
18-
go test -v ./...
18+
go test -v $(shell go list ./... | grep -v /e2e)
19+
20+
e2e-test:
21+
go test -v ./e2e/...
1922

2023
clean:
2124
rm -rf bin
25+
rm -rf ${BUNDLE_PATH}/../bundle.tar.gz ${BUNDLE_PATH} ${BUNDLE_PATH}/../bundle.tar.gz
Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
package append
1+
package add
22

33
import (
44
"errors"
@@ -20,7 +20,7 @@ func Flags() []cli.Flag {
2020

2121
func Command() *cli.Command {
2222
return &cli.Command{
23-
Name: "append",
23+
Name: "add",
2424
Flags: Flags(),
2525
Action: func(ctx *cli.Context) error {
2626
var (
@@ -93,6 +93,7 @@ func Command() *cli.Command {
9393
var tx = t.NewTransaction()
9494

9595
if err := tx.AddFiles(
96+
ctx.Context,
9697
lo.Map(inputFilesCol.Row(i), func(path string, _ int) string { return location.JoinPath("data", path).String() }),
9798
nil,
9899
true,

cmd/clickhouse/function/function.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
package function
22

33
import (
4-
"github.com/agnosticeng/icepq/cmd/clickhouse/function/append"
4+
"github.com/agnosticeng/icepq/cmd/clickhouse/function/add"
55
"github.com/agnosticeng/icepq/cmd/clickhouse/function/replace"
66
"github.com/urfave/cli/v2"
77
)
@@ -10,7 +10,7 @@ func Command() *cli.Command {
1010
return &cli.Command{
1111
Name: "function",
1212
Subcommands: []*cli.Command{
13-
append.Command(),
13+
add.Command(),
1414
replace.Command(),
1515
},
1616
}

cmd/clickhouse/function/replace/replace.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,7 @@ func Command() *cli.Command {
8888

8989
var tx = t.NewTransaction()
9090

91-
if err := tx.ReplaceDataFiles(inputFiles, outputFiles, nil); err != nil {
91+
if err := tx.ReplaceDataFiles(ctx.Context, inputFiles, outputFiles, nil); err != nil {
9292
return err
9393
}
9494

Lines changed: 15 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,21 +1,28 @@
1-
package append
1+
package add
22

33
import (
44
"errors"
5+
"fmt"
56
"net/url"
67

78
ice "github.com/agnosticeng/icepq/internal/iceberg"
8-
"github.com/apache/iceberg-go"
99
"github.com/apache/iceberg-go/catalog"
1010
"github.com/samber/lo"
1111
"github.com/urfave/cli/v2"
1212
)
1313

1414
func Command() *cli.Command {
1515
return &cli.Command{
16-
Name: "append",
17-
Usage: "append <location> <file1> [<file2> ...]",
16+
Name: "add",
17+
Usage: "add <location> <file1> [<file2> ...]",
18+
Flags: []cli.Flag{
19+
&cli.StringSliceFlag{Name: "prop"},
20+
},
1821
Action: func(ctx *cli.Context) error {
22+
var props = ice.ParseProperties(ctx.StringSlice("prop"))
23+
24+
fmt.Println(props)
25+
1926
location, err := url.Parse(ctx.Args().Get(0))
2027

2128
if err != nil {
@@ -34,7 +41,7 @@ func Command() *cli.Command {
3441
return err
3542
}
3643

37-
t, err := cat.LoadTable(ctx.Context, nil, iceberg.Properties{})
44+
t, err := cat.LoadTable(ctx.Context, nil, props)
3845

3946
if errors.Is(err, catalog.ErrNoSuchTable) {
4047
sch, err := ice.SchemaFromParquetDataFiles(ctx.Context, location, paths)
@@ -43,7 +50,7 @@ func Command() *cli.Command {
4350
return err
4451
}
4552

46-
t, err = cat.CreateTable(ctx.Context, nil, sch)
53+
t, err = cat.CreateTable(ctx.Context, nil, sch, catalog.WithProperties(props))
4754

4855
if err != nil {
4956
return err
@@ -57,8 +64,9 @@ func Command() *cli.Command {
5764
var tx = t.NewTransaction()
5865

5966
if err := tx.AddFiles(
67+
ctx.Context,
6068
lo.Map(paths, func(path string, _ int) string { return location.JoinPath("data", path).String() }),
61-
nil,
69+
props,
6270
true,
6371
); err != nil {
6472
return err

cmd/table/replace/replace.go

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@ import (
66
"strings"
77

88
ice "github.com/agnosticeng/icepq/internal/iceberg"
9-
"github.com/apache/iceberg-go"
109
"github.com/samber/lo"
1110
"github.com/urfave/cli/v2"
1211
)
@@ -15,7 +14,12 @@ func Command() *cli.Command {
1514
return &cli.Command{
1615
Name: "replace",
1716
Usage: "replace <location> <input_file_1,input_file_2,...> <output_file_1,output_file_2>",
17+
Flags: []cli.Flag{
18+
&cli.StringSliceFlag{Name: "prop"},
19+
},
1820
Action: func(ctx *cli.Context) error {
21+
var props = ice.ParseProperties(ctx.StringSlice("prop"))
22+
1923
location, err := url.Parse(ctx.Args().Get(0))
2024

2125
if err != nil {
@@ -28,7 +32,7 @@ func Command() *cli.Command {
2832
return err
2933
}
3034

31-
t, err := cat.LoadTable(ctx.Context, nil, iceberg.Properties{})
35+
t, err := cat.LoadTable(ctx.Context, nil, props)
3236

3337
if err != nil {
3438
return err
@@ -48,7 +52,7 @@ func Command() *cli.Command {
4852

4953
var tx = t.NewTransaction()
5054

51-
if err := tx.ReplaceDataFiles(inputFiles, outputFiles, nil); err != nil {
55+
if err := tx.ReplaceDataFiles(ctx.Context, inputFiles, outputFiles, props); err != nil {
5256
return err
5357
}
5458

cmd/table/table.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
package table
22

33
import (
4-
"github.com/agnosticeng/icepq/cmd/table/append"
4+
"github.com/agnosticeng/icepq/cmd/table/add"
55
"github.com/agnosticeng/icepq/cmd/table/files"
66
"github.com/agnosticeng/icepq/cmd/table/replace"
77
"github.com/urfave/cli/v2"
@@ -11,7 +11,7 @@ func Command() *cli.Command {
1111
return &cli.Command{
1212
Name: "table",
1313
Subcommands: []*cli.Command{
14-
append.Command(),
14+
add.Command(),
1515
files.Command(),
1616
replace.Command(),
1717
},
Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,10 @@
11
<functions>
22
<function>
3-
<name>icepq_append</name>
3+
<name>icepq_add</name>
44
<type>executable</type>
55
<format>Native</format>
66
<stderr_reaction>log</stderr_reaction>
7-
<command>icepq clickhouse function append</command>
7+
<command>icepq clickhouse function add</command>
88
<max_command_execution_time>300</max_command_execution_time>
99
<command_read_timeout>180000</command_read_timeout>
1010
<command_write_timeout>60000</command_write_timeout>

e2e/common/docker_setup.go

Lines changed: 163 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,163 @@
1+
package common
2+
3+
import (
4+
"context"
5+
_ "embed"
6+
"fmt"
7+
"os"
8+
"strings"
9+
"testing"
10+
"time"
11+
12+
"github.com/ClickHouse/clickhouse-go/v2"
13+
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
14+
"github.com/minio/minio-go/v7"
15+
"github.com/minio/minio-go/v7/pkg/credentials"
16+
"github.com/stretchr/testify/require"
17+
"github.com/testcontainers/testcontainers-go"
18+
miniotest "github.com/testcontainers/testcontainers-go/modules/minio"
19+
"github.com/testcontainers/testcontainers-go/network"
20+
)
21+
22+
type DockerSetup struct {
23+
Network *testcontainers.DockerNetwork
24+
Minio *miniotest.MinioContainer
25+
ClickHouse testcontainers.Container
26+
}
27+
28+
func (setup *DockerSetup) CreateClickhouseClient(ctx context.Context, t *testing.T) (driver.Conn, error) {
29+
clickhousPort, err := setup.ClickHouse.MappedPort(ctx, "9000")
30+
31+
if err != nil {
32+
return nil, err
33+
}
34+
35+
opts, err := clickhouse.ParseDSN(fmt.Sprintf("tcp://default:test@localhost:%d/default", clickhousPort.Int()))
36+
37+
if err != nil {
38+
return nil, err
39+
}
40+
41+
conn, err := clickhouse.Open(opts)
42+
43+
if err != nil {
44+
return nil, err
45+
}
46+
47+
pingCtx, cancel := context.WithTimeout(ctx, time.Second*10)
48+
defer cancel()
49+
50+
pingLoop:
51+
for {
52+
select {
53+
case <-pingCtx.Done():
54+
return nil, pingCtx.Err()
55+
case <-time.After(time.Second):
56+
t.Log("Trying to ping ClickHouse server...")
57+
err = conn.Ping(ctx)
58+
if err == nil {
59+
break pingLoop
60+
}
61+
return nil, err
62+
}
63+
}
64+
65+
ctx = clickhouse.Context(
66+
ctx,
67+
clickhouse.WithSettings(clickhouse.Settings{
68+
"send_logs_level": "debug",
69+
}),
70+
clickhouse.WithLogs(func(l *clickhouse.Log) {
71+
if strings.Contains(l.Text, "Executable generates stderr:") {
72+
t.Log(l.Text)
73+
}
74+
}),
75+
)
76+
77+
return conn, nil
78+
}
79+
80+
func WithDockerSetup(
81+
ctx context.Context,
82+
t *testing.T,
83+
bundlePath string,
84+
f func(*testing.T, *DockerSetup),
85+
) {
86+
net, err := network.New(ctx)
87+
require.NoError(t, err)
88+
testcontainers.CleanupNetwork(t, net)
89+
90+
// start minion container
91+
minioContainer, err := miniotest.Run(
92+
ctx,
93+
"minio/minio:RELEASE.2024-01-16T16-07-38Z",
94+
testcontainers.CustomizeRequest(testcontainers.GenericContainerRequest{
95+
ContainerRequest: testcontainers.ContainerRequest{
96+
Networks: []string{net.Name},
97+
NetworkAliases: map[string][]string{
98+
net.Name: {"minio"},
99+
},
100+
},
101+
}),
102+
miniotest.WithUsername("minio"),
103+
miniotest.WithPassword("minio123"),
104+
)
105+
require.NoError(t, err)
106+
defer testcontainers.TerminateContainer(minioContainer)
107+
108+
// create test bucket
109+
url, err := minioContainer.ConnectionString(ctx)
110+
require.NoError(t, err)
111+
minioClient, err := minio.New(url, &minio.Options{
112+
Creds: credentials.NewStaticV4(minioContainer.Username, minioContainer.Password, ""),
113+
Secure: false,
114+
})
115+
require.NoError(t, err)
116+
err = minioClient.MakeBucket(ctx, "test", minio.MakeBucketOptions{})
117+
require.NoError(t, err)
118+
119+
// start clickhouse-server container and mount UDF bundle
120+
bundleReader, err := os.Open(bundlePath)
121+
require.NoError(t, err)
122+
defer bundleReader.Close()
123+
clickhouseContainer, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
124+
ContainerRequest: testcontainers.ContainerRequest{
125+
Image: "clickhouse/clickhouse-server:25.3",
126+
Networks: []string{net.Name},
127+
NetworkAliases: map[string][]string{
128+
net.Name: {"clickhouse"},
129+
},
130+
Env: map[string]string{
131+
"CLICKHOUSE_PASSWORD": "test",
132+
"OBJSTR__S3__REGION": "us-east-1",
133+
"OBJSTR__S3__ACCESS_KEY_ID": "minio",
134+
"OBJSTR__S3__SECRET_ACCESS_KEY": "minio123",
135+
"OBJSTR__S3__DISABLE_SSL": "true",
136+
"OBJSTR__S3__FORCE_PATH_STYLE": "true",
137+
"OBJSTR__S3__ENDPOINT": "http://minio:9000",
138+
},
139+
Files: []testcontainers.ContainerFile{
140+
{
141+
Reader: bundleReader,
142+
HostFilePath: bundlePath,
143+
ContainerFilePath: "/bundle.tar.gz",
144+
FileMode: 700,
145+
},
146+
},
147+
},
148+
Started: true,
149+
})
150+
require.NoError(t, err)
151+
defer testcontainers.TerminateContainer(clickhouseContainer)
152+
153+
// install UDF bundle inside clickhouse-server container
154+
c, _, err := clickhouseContainer.Exec(ctx, []string{"tar", "-xvzf", "/bundle.tar.gz", "-C", "/"})
155+
require.NoError(t, err)
156+
require.Zero(t, c)
157+
158+
f(t, &DockerSetup{
159+
Network: net,
160+
Minio: minioContainer,
161+
ClickHouse: clickhouseContainer,
162+
})
163+
}

e2e/common/utils.go

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
package common
2+
3+
import (
4+
"context"
5+
"testing"
6+
7+
"github.com/stretchr/testify/require"
8+
"github.com/testcontainers/testcontainers-go"
9+
)
10+
11+
func ExecInContainer(ctx context.Context, t *testing.T, container testcontainers.Container, cmd []string) {
12+
c, _, err := container.Exec(ctx, []string{"pip", "install", "duckdb"})
13+
require.NoError(t, err)
14+
require.Zero(t, c)
15+
}

0 commit comments

Comments
 (0)