Skip to content

Commit b321fc7

Browse files
authored
Include hints in TableService direct API (#245)
* Include hints in TableService direct API * Add test
1 parent fd65243 commit b321fc7

2 files changed

Lines changed: 36 additions & 1 deletion

File tree

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/TableService.java

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99

1010
import java.sql.SQLException;
1111
import java.util.Collection;
12+
import java.util.HashMap;
1213
import java.util.List;
1314
import java.util.Map;
1415
import java.util.Properties;
@@ -81,8 +82,12 @@ public static HoptimatorDdlUtils.SpecifyResult create(Properties connectionPrope
8182
String tableName = path.get(path.size() - 1);
8283
DirectDeploymentContext context = new DirectDeploymentContext(connectionProperties, resolver, avroSchema);
8384

85+
// Order is intentional to prevent callers from overriding connection properties leading to impersonation
86+
Map<String, String> tableOptions = new HashMap<>(options);
87+
tableOptions.putAll(DeploymentService.parseHints(connectionProperties));
88+
8489
return HoptimatorDdlUtils.deployTableInternal(logHooks, context, null, path,
85-
database, tableName, options, false, updateIfExists, mode);
90+
database, tableName, tableOptions, false, updateIfExists, mode);
8691
}
8792

8893
/**

hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/TableServiceTest.java

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import org.apache.calcite.sql.type.SqlTypeName;
1010
import org.junit.jupiter.api.Test;
1111
import org.junit.jupiter.api.extension.ExtendWith;
12+
import org.mockito.ArgumentCaptor;
1213
import org.mockito.Mock;
1314
import org.mockito.MockedStatic;
1415
import org.mockito.junit.jupiter.MockitoExtension;
@@ -17,6 +18,7 @@
1718
import java.sql.SQLNonTransientException;
1819
import java.util.Collections;
1920
import java.util.List;
21+
import java.util.Map;
2022
import java.util.Properties;
2123

2224
import static org.assertj.core.api.Assertions.assertThat;
@@ -187,6 +189,34 @@ void updateIfExistsBypassesTheGuardAndDoesNotConsultExists() throws SQLException
187189
verify(deployer, never()).exists();
188190
}
189191

192+
@Test
193+
void createMergesConnectionHintsIntoTableOptionsWithHintsWinning() throws SQLException {
194+
// Connection hints must be merged into the table options handed to the deployers, and a hint
195+
// must override a caller-supplied option of the same key so callers can't override
196+
// connection-level properties (impersonation guard).
197+
DatabaseConfigResolver resolver = stubResolver();
198+
resolvers.when(() -> DatabaseConfigResolvers.forProperties(any())).thenReturn(resolver);
199+
deployment.when(() -> DeploymentService.parseHints(any()))
200+
.thenReturn(Map.of("owner", "connection-user", "hintOnly", "hintValue"));
201+
Deployer deployer = mock(Deployer.class);
202+
when(deployer.exists()).thenReturn(false);
203+
List<Deployer> deployers = Collections.singletonList(deployer);
204+
deployment.when(() -> DeploymentService.deployers(any(Source.class), any(DeploymentContext.class)))
205+
.thenReturn(deployers);
206+
207+
Map<String, String> callerOptions = Map.of("owner", "caller-attempt", "callerOnly", "callerValue");
208+
TableService.create(new Properties(), Collections.emptyList(), path, recordSchema(),
209+
callerOptions, false, false);
210+
211+
ArgumentCaptor<Source> sourceCaptor = ArgumentCaptor.forClass(Source.class);
212+
deployment.verify(() ->
213+
DeploymentService.deployers(sourceCaptor.capture(), any(DeploymentContext.class)));
214+
Map<String, String> mergedOptions = sourceCaptor.getValue().options();
215+
assertThat(mergedOptions).containsEntry("callerOnly", "callerValue");
216+
assertThat(mergedOptions).containsEntry("hintOnly", "hintValue");
217+
assertThat(mergedOptions).containsEntry("owner", "connection-user");
218+
}
219+
190220
private static Schema recordSchema() {
191221
return new Schema.Parser().parse(RECORD_SCHEMA);
192222
}

0 commit comments

Comments
 (0)