Skip to content

Commit 10a568b

Browse files
miniwayminiway
committed
Make new Iceberg plugin compile
1 parent a2f7a5b commit 10a568b

16 files changed

Lines changed: 448 additions & 33 deletions

File tree

‎lib/trino-hive-formats/pom.xml‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -194,12 +194,14 @@
194194
<scope>test</scope>
195195
</dependency>
196196

197+
<!--
197198
<dependency>
198199
<groupId>org.apache.commons</groupId>
199200
<artifactId>commons-lang3</artifactId>
200-
<version>3.12.0</version>
201+
<version>3.18.0</version>
201202
<scope>test</scope>
202203
</dependency>
204+
-->
203205

204206
<dependency>
205207
<groupId>org.assertj</groupId>

‎lib/trino-parquet/pom.xml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,12 @@
8484
<dependency>
8585
<groupId>org.apache.parquet</groupId>
8686
<artifactId>parquet-format-structures</artifactId>
87+
<exclusions>
88+
<exclusion>
89+
<groupId>javax.annotation</groupId>
90+
<artifactId>javax.annotation-api</artifactId>
91+
</exclusion>
92+
</exclusions>
8793
</dependency>
8894

8995
<dependency>

‎plugin/trino-bigquery/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@
3636
<dependency>
3737
<groupId>org.apache.commons</groupId>
3838
<artifactId>commons-lang3</artifactId>
39-
<version>3.11</version>
39+
<version>3.18.0</version>
4040
</dependency>
4141

4242
<dependency>

‎plugin/trino-delta-lake/pom.xml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -192,6 +192,12 @@
192192
<dependency>
193193
<groupId>org.apache.parquet</groupId>
194194
<artifactId>parquet-format-structures</artifactId>
195+
<exclusions>
196+
<exclusion>
197+
<groupId>javax.annotation</groupId>
198+
<artifactId>javax.annotation-api</artifactId>
199+
</exclusion>
200+
</exclusions>
195201
</dependency>
196202

197203
<dependency>

‎plugin/trino-hive/pom.xml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -259,6 +259,12 @@
259259
<dependency>
260260
<groupId>org.apache.parquet</groupId>
261261
<artifactId>parquet-format-structures</artifactId>
262+
<exclusions>
263+
<exclusion>
264+
<groupId>javax.annotation</groupId>
265+
<artifactId>javax.annotation-api</artifactId>
266+
</exclusion>
267+
</exclusions>
262268
</dependency>
263269

264270
<dependency>

‎plugin/trino-iceberg/pom.xml‎

Lines changed: 20 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
-->
2626
<air.test.parallel>instances</air.test.parallel>
2727
<!-- Nessie version (matching to Iceberg release) must be bumped along with Iceberg version bump to avoid compatibility issues -->
28-
<dep.nessie.version>0.59.0</dep.nessie.version>
28+
<dep.nessie.version>0.104.4</dep.nessie.version>
2929
</properties>
3030

3131
<dependencies>
@@ -213,6 +213,17 @@
213213
<artifactId>iceberg-api</artifactId>
214214
</dependency>
215215

216+
<dependency>
217+
<groupId>org.apache.iceberg</groupId>
218+
<artifactId>iceberg-aws</artifactId>
219+
<exclusions>
220+
<exclusion>
221+
<groupId>org.jspecify</groupId>
222+
<artifactId>jspecify</artifactId>
223+
</exclusion>
224+
</exclusions>
225+
</dependency>
226+
216227
<dependency>
217228
<groupId>org.apache.iceberg</groupId>
218229
<artifactId>iceberg-core</artifactId>
@@ -247,6 +258,12 @@
247258
<dependency>
248259
<groupId>org.apache.parquet</groupId>
249260
<artifactId>parquet-format-structures</artifactId>
261+
<exclusions>
262+
<exclusion>
263+
<groupId>javax.annotation</groupId>
264+
<artifactId>javax.annotation-api</artifactId>
265+
</exclusion>
266+
</exclusions>
250267
</dependency>
251268

252269
<dependency>
@@ -368,14 +385,14 @@
368385
<dependency>
369386
<groupId>org.apache.httpcomponents.client5</groupId>
370387
<artifactId>httpclient5</artifactId>
371-
<version>5.2.1</version>
388+
<version>5.4.3</version>
372389
<scope>runtime</scope>
373390
</dependency>
374391

375392
<dependency>
376393
<groupId>org.apache.httpcomponents.core5</groupId>
377394
<artifactId>httpcore5</artifactId>
378-
<version>5.2.1</version>
395+
<version>5.3.4</version>
379396
<scope>runtime</scope>
380397
</dependency>
381398

‎plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/nessie/IcebergNessieCatalogConfig.java‎

Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,16 +15,48 @@
1515

1616
import io.airlift.configuration.Config;
1717
import io.airlift.configuration.ConfigDescription;
18+
import io.airlift.configuration.ConfigSecuritySensitive;
19+
import io.airlift.units.Duration;
20+
import io.airlift.units.MinDuration;
21+
import jakarta.validation.constraints.AssertTrue;
1822
import jakarta.validation.constraints.NotEmpty;
1923
import jakarta.validation.constraints.NotNull;
2024

2125
import java.net.URI;
26+
import java.util.Optional;
27+
import java.util.regex.Matcher;
28+
import java.util.regex.Pattern;
29+
30+
import static com.google.common.base.Preconditions.checkArgument;
31+
import static io.trino.plugin.iceberg.catalog.nessie.IcebergNessieCatalogConfig.Security.BEARER;
32+
import static java.util.concurrent.TimeUnit.MILLISECONDS;
33+
import static java.util.function.Predicate.isEqual;
34+
import static org.projectnessie.client.NessieConfigConstants.DEFAULT_CONNECT_TIMEOUT_MILLIS;
35+
import static org.projectnessie.client.NessieConfigConstants.DEFAULT_READ_TIMEOUT_MILLIS;
2236

2337
public class IcebergNessieCatalogConfig
2438
{
39+
public enum Security
40+
{
41+
BEARER,
42+
}
43+
44+
public enum ClientApiVersion
45+
{
46+
V1,
47+
V2,
48+
}
49+
2550
private String defaultReferenceName = "main";
2651
private String defaultWarehouseDir;
2752
private URI serverUri;
53+
private Duration readTimeout = new Duration(DEFAULT_READ_TIMEOUT_MILLIS, MILLISECONDS);
54+
private Duration connectionTimeout = new Duration(DEFAULT_CONNECT_TIMEOUT_MILLIS, MILLISECONDS);
55+
private boolean enableCompression = true;
56+
private Security security;
57+
private Optional<String> bearerToken = Optional.empty();
58+
private Optional<ClientApiVersion> clientAPIVersion = Optional.empty();
59+
private static final Pattern VERSION_PATTERN = Pattern.compile("/v(\\d+)$");
2860

2961
@NotNull
3062
public String getDefaultReferenceName()
@@ -67,4 +99,113 @@ public IcebergNessieCatalogConfig setDefaultWarehouseDir(String defaultWarehouse
6799
this.defaultWarehouseDir = defaultWarehouseDir;
68100
return this;
69101
}
102+
103+
@MinDuration("1ms")
104+
public Duration getReadTimeout()
105+
{
106+
return readTimeout;
107+
}
108+
109+
@Config("iceberg.nessie-catalog.read-timeout")
110+
@ConfigDescription("The read timeout for the client.")
111+
public IcebergNessieCatalogConfig setReadTimeout(Duration readTimeout)
112+
{
113+
this.readTimeout = readTimeout;
114+
return this;
115+
}
116+
117+
@MinDuration("1ms")
118+
public Duration getConnectionTimeout()
119+
{
120+
return connectionTimeout;
121+
}
122+
123+
@Config("iceberg.nessie-catalog.connection-timeout")
124+
@ConfigDescription("The connection timeout for the client.")
125+
public IcebergNessieCatalogConfig setConnectionTimeout(Duration connectionTimeout)
126+
{
127+
this.connectionTimeout = connectionTimeout;
128+
return this;
129+
}
130+
131+
public boolean isCompressionEnabled()
132+
{
133+
return enableCompression;
134+
}
135+
136+
@Config("iceberg.nessie-catalog.enable-compression")
137+
@ConfigDescription("Configure whether compression should be enabled or not.")
138+
public IcebergNessieCatalogConfig setCompressionEnabled(boolean enableCompression)
139+
{
140+
this.enableCompression = enableCompression;
141+
return this;
142+
}
143+
144+
public Optional<Security> getSecurity()
145+
{
146+
return Optional.ofNullable(security);
147+
}
148+
149+
@Config("iceberg.nessie-catalog.authentication.type")
150+
@ConfigDescription("The authentication type to use")
151+
public IcebergNessieCatalogConfig setSecurity(Security security)
152+
{
153+
this.security = security;
154+
return this;
155+
}
156+
157+
public Optional<String> getBearerToken()
158+
{
159+
return bearerToken;
160+
}
161+
162+
@Config("iceberg.nessie-catalog.authentication.token")
163+
@ConfigDescription("The token to use with BEARER authentication")
164+
@ConfigSecuritySensitive
165+
public IcebergNessieCatalogConfig setBearerToken(String token)
166+
{
167+
this.bearerToken = Optional.ofNullable(token);
168+
return this;
169+
}
170+
171+
@AssertTrue(message = "'iceberg.nessie-catalog.authentication.token' must be configured only with 'iceberg.nessie-catalog.authentication.type' BEARER")
172+
public boolean isTokenConfiguredWithoutType()
173+
{
174+
return getSecurity().filter(isEqual(BEARER)).isPresent() || getBearerToken().isEmpty();
175+
}
176+
177+
@AssertTrue(message = "'iceberg.nessie-catalog.authentication.token' must be configured with 'iceberg.nessie-catalog.authentication.type' BEARER")
178+
public boolean isMissingTokenForBearerAuth()
179+
{
180+
return getSecurity().filter(isEqual(BEARER)).isEmpty() || getBearerToken().isPresent();
181+
}
182+
183+
public Optional<ClientApiVersion> getClientAPIVersion()
184+
{
185+
return clientAPIVersion;
186+
}
187+
188+
@Config("iceberg.nessie-catalog.client-api-version")
189+
@ConfigDescription("Client API version to use")
190+
public IcebergNessieCatalogConfig setClientAPIVersion(ClientApiVersion version)
191+
{
192+
this.clientAPIVersion = Optional.ofNullable(version);
193+
return this;
194+
}
195+
196+
protected IcebergNessieCatalogConfig.ClientApiVersion inferVersionFromURI()
197+
{
198+
checkArgument(serverUri != null, "URI is not specified in the catalog properties");
199+
// match for uri ending with /v1, /v2 etc
200+
Matcher matcher = VERSION_PATTERN.matcher(serverUri.toString());
201+
if (!matcher.find()) {
202+
throw new IllegalArgumentException("URI doesn't end with the version: %s. Please configure `client-api-version` in the catalog properties explicitly.".formatted(serverUri));
203+
}
204+
205+
return switch (matcher.group(1)) {
206+
case "1" -> ClientApiVersion.V1;
207+
case "2" -> ClientApiVersion.V2;
208+
default -> throw new IllegalArgumentException("Unknown API version in the URI: " + matcher.group(1));
209+
};
210+
}
70211
}

‎plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/nessie/IcebergNessieCatalogModule.java‎

Lines changed: 21 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -22,10 +22,13 @@
2222
import io.trino.plugin.iceberg.catalog.IcebergTableOperationsProvider;
2323
import io.trino.plugin.iceberg.catalog.TrinoCatalogFactory;
2424
import org.apache.iceberg.nessie.NessieIcebergClient;
25+
import org.projectnessie.client.NessieClientBuilder;
2526
import org.projectnessie.client.api.NessieApiV1;
26-
import org.projectnessie.client.http.HttpClientBuilder;
27+
import org.projectnessie.client.api.NessieApiV2;
28+
import org.projectnessie.client.auth.BearerAuthenticationProvider;
2729

2830
import static io.airlift.configuration.ConfigBinder.configBinder;
31+
import static java.lang.Math.toIntExact;
2932
import static org.weakref.jmx.guice.ExportBinder.newExporter;
3033

3134
public class IcebergNessieCatalogModule
@@ -45,11 +48,23 @@ protected void setup(Binder binder)
4548
@Singleton
4649
public static NessieIcebergClient createNessieIcebergClient(IcebergNessieCatalogConfig icebergNessieCatalogConfig)
4750
{
48-
return new NessieIcebergClient(
49-
HttpClientBuilder.builder()
50-
.withUri(icebergNessieCatalogConfig.getServerUri())
51-
.withEnableApiCompatibilityCheck(false)
52-
.build(NessieApiV1.class),
51+
NessieClientBuilder builder = NessieClientBuilder.createClientBuilderFromSystemSettings()
52+
.withUri(icebergNessieCatalogConfig.getServerUri())
53+
.withDisableCompression(!icebergNessieCatalogConfig.isCompressionEnabled())
54+
.withReadTimeout(toIntExact(icebergNessieCatalogConfig.getReadTimeout().toMillis()))
55+
.withConnectionTimeout(toIntExact(icebergNessieCatalogConfig.getConnectionTimeout().toMillis()));
56+
57+
icebergNessieCatalogConfig.getBearerToken()
58+
.ifPresent(token -> builder.withAuthentication(BearerAuthenticationProvider.create(token)));
59+
60+
IcebergNessieCatalogConfig.ClientApiVersion clientApiVersion = icebergNessieCatalogConfig.getClientAPIVersion()
61+
.orElseGet(icebergNessieCatalogConfig::inferVersionFromURI);
62+
NessieApiV1 api = switch (clientApiVersion) {
63+
case V1 -> builder.build(NessieApiV1.class);
64+
case V2 -> builder.build(NessieApiV2.class);
65+
};
66+
67+
return new NessieIcebergClient(api,
5368
icebergNessieCatalogConfig.getDefaultReferenceName(),
5469
null,
5570
ImmutableMap.of());

‎plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/nessie/IcebergNessieTableOperations.java‎

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -96,8 +96,14 @@ protected String getRefreshedLocation(boolean invalidateCaches)
9696
protected void commitNewTable(TableMetadata metadata)
9797
{
9898
verify(version.isEmpty(), "commitNewTable called on a table which already exists");
99+
String contentId = table == null ? null : table.getId();
99100
try {
100-
nessieClient.commitTable(null, metadata, writeNewMetadata(metadata, 0), table, toKey(new SchemaTableName(database, this.tableName)));
101+
nessieClient.commitTable(
102+
null,
103+
metadata,
104+
writeNewMetadata(metadata, 0),
105+
contentId,
106+
toKey(database, tableName));
101107
}
102108
catch (NessieNotFoundException e) {
103109
throw new TrinoException(ICEBERG_COMMIT_ERROR, format("Cannot commit: ref '%s' no longer exists", nessieClient.refName()), e);
@@ -113,8 +119,16 @@ protected void commitNewTable(TableMetadata metadata)
113119
protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)
114120
{
115121
verify(version.orElseThrow() >= 0, "commitToExistingTable called on a new table");
122+
if (table == null) {
123+
table = nessieClient.table(toIdentifier(new SchemaTableName(database, tableName)));
124+
}
116125
try {
117-
nessieClient.commitTable(base, metadata, writeNewMetadata(metadata, version.getAsInt() + 1), table, toKey(new SchemaTableName(database, this.tableName)));
126+
nessieClient.commitTable(
127+
base,
128+
metadata,
129+
writeNewMetadata(metadata, version.getAsInt() + 1),
130+
table.getId(),
131+
toKey(database, tableName));
118132
}
119133
catch (NessieNotFoundException e) {
120134
throw new TrinoException(ICEBERG_COMMIT_ERROR, format("Cannot commit: ref '%s' no longer exists", nessieClient.refName()), e);
@@ -126,8 +140,9 @@ protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)
126140
shouldRefresh = true;
127141
}
128142

129-
private static ContentKey toKey(SchemaTableName tableName)
143+
private static ContentKey toKey(String databaseName, String tableName)
130144
{
131-
return ContentKey.of(Namespace.parse(tableName.getSchemaName()), tableName.getTableName());
145+
SchemaTableName schemaTableName = new SchemaTableName(databaseName, tableName);
146+
return ContentKey.of(Namespace.parse(schemaTableName.getSchemaName()), schemaTableName.getTableName());
132147
}
133148
}

0 commit comments

Comments
 (0)