Extracts the field path and comparison value from expressions like: + *
This class centralizes all Accumulo client creation and authentication logic, + * supporting:
+ *For PASSWORD mode, this uses the configured username/password. + * For KERBEROS mode, this authenticates using the service principal and keytab.
+ * + *The client is lazily initialized and cached for reuse.
+ * + * @return the service AccumuloClient + * @throws UserException if connection fails + */ + public AccumuloClient getServiceClient() { + if (serviceClient == null) { + synchronized (serviceClientLock) { + if (serviceClient == null) { + try { + serviceClient = createServiceClient(); + logger.info("Created Accumulo service client for instance: {}", config.getInstanceName()); + } catch (AccumuloException | AccumuloSecurityException | IOException e) { + throw UserException.connectionError(e) + .message("Failed to connect to Accumulo instance '%s' at '%s'", + config.getInstanceName(), config.getZookeeperQuorum()) + .addContext("AuthenticationType", config.getAuthenticationType()) + .build(logger); + } + } + } + } + return serviceClient; + } + + /** + * Returns an AccumuloClient for the specified user. + * + *Behavior depends on the auth mode:
+ *In this mode, per-user Accumulo credentials are looked up from the + * CredentialsProvider based on the Drill query user.
+ */ + private AccumuloClient getClientForUserTranslation(String userName, UserCredentials userCredentials) { + // Build UserCredentials if not provided + if (userCredentials == null && userName != null) { + userCredentials = UserCredentials.newBuilder() + .setUserName(userName) + .build(); + } + + // Look up per-user credentials + OptionalTokens are cached with a TTL to avoid repeated token creation. + * This method is thread-safe.
+ * + * @param userName the user to get a delegation token for + * @return the delegation token info + * @throws AccumuloException if token creation fails + * @throws AccumuloSecurityException if authentication fails + * @throws IOException if token serialization fails + */ + public DelegationTokenInfo getDelegationToken(String userName) + throws AccumuloException, AccumuloSecurityException, IOException { + + // Check cache first + DelegationTokenInfo cached = delegationTokenCache.get(userName); + if (cached != null && !cached.isOlderThan(DEFAULT_TOKEN_TTL_MILLIS)) { + logger.debug("Using cached delegation token for user: {}", userName); + return cached; + } + + // Need to create/refresh token + synchronized (delegationTokenCache) { + // Double-check after acquiring lock + cached = delegationTokenCache.get(userName); + if (cached != null && !cached.isOlderThan(DEFAULT_TOKEN_TTL_MILLIS)) { + return cached; + } + + logger.info("Creating delegation token for user: {}", userName); + + // Use the service client to obtain a delegation token + AccumuloClient client = getServiceClient(); + + // Create a proxy user UGI for the query user + UserGroupInformation proxyUgi = ImpersonationUtil.createProxyUgi(userName); + + // Request delegation token for the proxy user + DelegationTokenConfig tokenConfig = new DelegationTokenConfig(); + tokenConfig.setTokenLifetime(DEFAULT_TOKEN_LIFETIME_MILLIS, TimeUnit.MILLISECONDS); + + DelegationToken token = client.securityOperations() + .getDelegationToken(tokenConfig); + + DelegationTokenInfo tokenInfo = DelegationTokenInfo.fromDelegationToken(userName, token); + delegationTokenCache.put(userName, tokenInfo); + + logger.info("Created delegation token for user: {} (expires in {} ms)", + userName, DEFAULT_TOKEN_LIFETIME_MILLIS); + + return tokenInfo; + } + } + + /** + * Creates an AccumuloClient using a delegation token. + * + *This method is used by distributed fragments to create a client + * with the delegated user identity.
+ * + * @param tokenInfo the delegation token info + * @return a new AccumuloClient authenticated with the delegation token + * @throws AccumuloException if client creation fails + * @throws AccumuloSecurityException if authentication fails + */ + public AccumuloClient createClientWithDelegationToken(DelegationTokenInfo tokenInfo) + throws AccumuloException, AccumuloSecurityException { + + AuthenticationToken token = tokenInfo.toAuthenticationToken(); + + Properties props = new Properties(); + props.setProperty("instance.name", config.getInstanceName()); + props.setProperty("instance.zookeepers", config.getZookeeperQuorum()); + + // Configure SASL for delegation token authentication + if (!Strings.isNullOrEmpty(config.getSaslQop())) { + props.setProperty("rpc.sasl.qop", config.getSaslQop()); + } + + return Accumulo.newClient() + .from(props) + .as(tokenInfo.getUserName(), token) + .build(); + } + + /** + * Creates an AccumuloClient using username/password credentials. + * + *This method is used for USER_TRANSLATION mode where per-user + * credentials are stored in the CredentialsProvider.
+ * + * @param credentials the username/password credentials + * @return a new AccumuloClient + * @throws AccumuloException if client creation fails + * @throws AccumuloSecurityException if authentication fails + */ + public AccumuloClient createClientWithCredentials(UsernamePasswordCredentials credentials) + throws AccumuloException, AccumuloSecurityException { + + Properties props = new Properties(); + props.setProperty("instance.name", config.getInstanceName()); + props.setProperty("instance.zookeepers", config.getZookeeperQuorum()); + + return Accumulo.newClient() + .from(props) + .as(credentials.getUsername(), new PasswordToken(credentials.getPassword())) + .build(); + } + + /** + * Creates the service client based on the configured authentication type. + */ + private AccumuloClient createServiceClient() + throws AccumuloException, AccumuloSecurityException, IOException { + + AccumuloAuthType authType = config.getAuthenticationType(); + + Properties props = new Properties(); + props.setProperty("instance.name", config.getInstanceName()); + props.setProperty("instance.zookeepers", config.getZookeeperQuorum()); + + String principal; + AuthenticationToken token; + + if (authType == AccumuloAuthType.KERBEROS) { + principal = config.getPrincipal(); + token = createKerberosToken(); + + // Configure SASL properties + if (!Strings.isNullOrEmpty(config.getSaslQop())) { + props.setProperty("rpc.sasl.qop", config.getSaslQop()); + } + if (!Strings.isNullOrEmpty(config.getAccumuloServicePrimary())) { + props.setProperty("sasl.kerberos.server.primary", config.getAccumuloServicePrimary()); + } + } else { + // PASSWORD authentication + OptionalThis class converts Drill's LogicalExpression filter representation into + * Accumulo scan parameters (start row, stop row). It focuses on row key + * predicates since those can be efficiently pushed down to Accumulo's scan range.
+ * + *Supported predicates on row_key:
+ *An AND of row key predicates is an exact translation, so the Drill filter can be + * removed once it is pushed down. An OR is not: the spanning range also covers the rows + * between the operands, so the range is treated as a partial pushdown and the filter is + * left in the plan to remove the extra rows.
+ */ +public class AccumuloFilterBuilder + extends AbstractExprVisitorThis class handles scan planning and fragmentation across Accumulo tablets. + * It can be modified by optimizer rules to apply filter, projection, limit, and sort pushdowns.
+ * + *For user impersonation mode, this class carries a delegation token that is + * serialized to JSON for distributed planning and passed to SubScans for execution.
+ */ +@JsonTypeName("accumulo-scan") +public class AccumuloGroupScan extends AbstractGroupScan { + + private AccumuloStoragePluginConfig storagePluginConfig; + private AccumuloStoragePlugin storagePlugin; + private AccumuloScanSpec scanSpec; + private ListThis rule matches Filter → Scan patterns (optionally with a Project in between) + * and pushes row_key predicates down to the Accumulo scan as row ranges.
+ * + *Supported patterns:
+ *Accumulo naturally returns rows sorted by row key in ascending order. + * This rule detects when a sort is on the row_key column and can be satisfied + * by Accumulo's natural ordering:
+ * + *When sort is pushed down, the sort operator may be eliminated by + * subsequent optimization passes since the data is already sorted.
+ */ +public abstract class AccumuloPushSortIntoScan extends StoragePluginOptimizerRule { + private static final Logger logger = LoggerFactory.getLogger(AccumuloPushSortIntoScan.class); + + private AccumuloPushSortIntoScan(RelOptRuleOperand operand, String description) { + super(operand, description); + } + + /** + * Rule for Sort directly on Scan. + */ + public static final StoragePluginOptimizerRule SORT_ON_SCAN = + new AccumuloPushSortIntoScan( + RelOptHelper.some(DrillSortRel.class, RelOptHelper.any(DrillScanRel.class)), + "AccumuloPushSortIntoScan:Sort_On_Scan") { + + @Override + public void onMatch(RelOptRuleCall call) { + DrillSortRel sort = call.rel(0); + DrillScanRel scan = call.rel(1); + doPushSortIntoScan(call, sort, scan); + } + + @Override + public boolean matches(RelOptRuleCall call) { + DrillScanRel scan = call.rel(1); + if (!(scan.getGroupScan() instanceof AccumuloGroupScan)) { + return false; + } + AccumuloGroupScan groupScan = (AccumuloGroupScan) scan.getGroupScan(); + // Don't push sort if already pushed + return !groupScan.isSortPushedDown(); + } + }; + + /** + * Pushes sort into Accumulo scan if the sort is on row_key. + */ + protected void doPushSortIntoScan(RelOptRuleCall call, DrillSortRel sort, DrillScanRel scan) { + AccumuloGroupScan groupScan = (AccumuloGroupScan) scan.getGroupScan(); + + // Check if sort is on row_key + RelCollation collation = sort.getCollation(); + ListThis reader scans Accumulo tables and populates Drill value vectors. + * It uses the dynamic schema approach similar to HBase, where column families + * are represented as maps containing their qualifiers as fields.
+ * + *Row structure:
+ *For user impersonation mode, the reader may own the AccumuloClient + * (created from a delegation token) and is responsible for closing it. + * For shared user mode, the reader uses a shared client and should not close it.
+ */ +public class AccumuloRecordReader extends AbstractRecordReader implements DrillAccumuloConstants { + private static final Logger logger = LoggerFactory.getLogger(AccumuloRecordReader.class); + + // Batch constraints to avoid OOM + private static final int MAX_ALLOCATED_MEMORY_PER_BATCH = 64 * 1024 * 1024; // 64 MB + private static final int TARGET_RECORD_COUNT = DEFAULT_BATCH_SIZE; + + private final AccumuloClient client; + private final AccumuloScanSpec scanSpec; + private final int maxRecords; + + /** + * Whether this reader owns the client and should close it. + * True for user impersonation mode (client created from delegation token), + * false for shared user mode (client is shared/pooled). + */ + private final boolean ownsClient; + + private OutputMutator outputMutator; + private OperatorContext operatorContext; + + private Scanner scanner; + private IteratorThis class creates the execution pipeline for Accumulo scans by wiring + * together the AccumuloSubScan with AccumuloRecordReaders.
+ * + *For user impersonation mode, this class creates clients using the + * delegation token passed from the SubScan. When a delegation token is + * present, the reader owns the client and is responsible for closing it.
+ */ +public class AccumuloScanBatchCreator implements BatchCreatorThis class captures all scan parameters that may be pushed down to Accumulo, + * including table name, row key ranges, column projections, filters, and limits.
+ */ +public class AccumuloScanSpec implements DrillTableSelection { + + private final String tableName; + private final byte[] startRow; + private final byte[] stopRow; + private final boolean startRowInclusive; + private final boolean stopRowInclusive; + private final ListResponsible for registering the Accumulo schema and discovering tables.
+ */ +public class AccumuloSchemaFactory extends AbstractSchemaFactory { + private static final Logger logger = LoggerFactory.getLogger(AccumuloSchemaFactory.class); + + private final AccumuloStoragePlugin plugin; + + public AccumuloSchemaFactory(AccumuloStoragePlugin plugin) { + super(plugin.getName()); + this.plugin = plugin; + } + + @Override + public void registerSchemas(SchemaConfig schemaConfig, SchemaPlus parent) throws IOException { + AccumuloSchema schema = new AccumuloSchema(getName()); + SchemaPlus schemaPlus = parent.add(getName(), schema); + schema.setHolder(schemaPlus); + } + + /** + * Accumulo schema implementation. + */ + class AccumuloSchema extends AbstractSchema { + + AccumuloSchema(String name) { + super(Collections.emptyList(), name); + } + + public void setHolder(SchemaPlus plusOfThis) { + // No-op for now + } + + @Override + public AbstractSchema getSubSchema(String name) { + return null; + } + + @Override + public SetThis plugin provides read access to Apache Accumulo tables, + * with support for filter, projection, limit, and sort pushdowns.
+ * + *Authentication modes:
+ *This is the service client that authenticates using the configured + * credentials (password or Kerberos). For user impersonation, use + * {@link #getClientForUser(String)} instead.
+ * + * @return the service Accumulo client + * @throws UserException if connection fails + */ + public AccumuloClient getClient() { + return connectionManager.getServiceClient(); + } + + /** + * Returns an AccumuloClient for the specified user. + * + *Behavior depends on the auth mode:
+ *This is used in distributed execution to pass the user's credentials + * to executor fragments.
+ * + * @param userName the user to generate a token for + * @return the delegation token info, or null if impersonation is not enabled + */ + public DelegationTokenInfo generateDelegationToken(String userName) { + if (!config.isUserImpersonationEnabled() || !config.isUseDelegationTokens()) { + return null; + } + + try { + return connectionManager.getDelegationToken(userName); + } catch (Exception e) { + throw UserException.connectionError(e) + .message("Failed to generate delegation token for user '%s'", userName) + .addContext("Plugin", getName()) + .build(logger); + } + } + + @Override + public AbstractGroupScan getPhysicalScan(String userName, JSONOptions selection) throws IOException { + AccumuloScanSpec scanSpec = selection.getListWith(new TypeReferenceThis configuration supports connecting to an Accumulo cluster via ZooKeeper + * and includes settings for authentication, Kerberos, user impersonation, + * and optional schema metadata table.
+ * + *
+ * {
+ * "type": "accumulo",
+ * "zookeeperQuorum": "localhost:2181",
+ * "instanceName": "accumulo",
+ * "username": "root",
+ * "password": "secret",
+ * "enabled": true
+ * }
+ *
+ *
+ *
+ * {
+ * "type": "accumulo",
+ * "zookeeperQuorum": "zk1:2181,zk2:2181",
+ * "instanceName": "accumulo",
+ * "authenticationType": "KERBEROS",
+ * "principal": "drill/hostname@REALM",
+ * "keytabPath": "/etc/security/keytabs/drill.keytab",
+ * "saslQop": "auth",
+ * "useDelegationTokens": true,
+ * "authMode": "USER_IMPERSONATION",
+ * "enabled": true
+ * }
+ *
+ */
+@JsonTypeName(AccumuloStoragePluginConfig.NAME)
+public class AccumuloStoragePluginConfig extends StoragePluginConfig {
+
+ public static final String NAME = "accumulo";
+
+ /**
+ * Default SASL QoP (Quality of Protection) value.
+ */
+ public static final String DEFAULT_SASL_QOP = "auth";
+
+ /**
+ * Default Accumulo service name for SASL authentication.
+ */
+ public static final String DEFAULT_ACCUMULO_SERVICE_PRIMARY = "accumulo";
+
+ // ===== Connection settings =====
+
+ /**
+ * Comma-separated list of ZooKeeper servers (host:port format).
+ * Example: "zk1:2181,zk2:2181,zk3:2181"
+ */
+ private final String zookeeperQuorum;
+
+ /**
+ * The Accumulo instance name.
+ */
+ private final String instanceName;
+
+ // ===== Password authentication settings (for backward compatibility) =====
+
+ /**
+ * The username for password authentication.
+ * Deprecated: prefer using credentialsProvider.
+ */
+ private final String username;
+
+ /**
+ * The password for password authentication.
+ * Deprecated: prefer using credentialsProvider.
+ */
+ private final String password;
+
+ // ===== Kerberos authentication settings =====
+
+ /**
+ * The authentication type: PASSWORD or KERBEROS.
+ */
+ private final AccumuloAuthType authenticationType;
+
+ /**
+ * Kerberos principal for service authentication.
+ * Format: primary/instance@REALM (e.g., "drill/hostname@EXAMPLE.COM")
+ */
+ private final String principal;
+
+ /**
+ * Path to the Kerberos keytab file.
+ */
+ private final String keytabPath;
+
+ /**
+ * SASL Quality of Protection: "auth", "auth-int", or "auth-conf".
+ * - auth: authentication only
+ * - auth-int: authentication + integrity protection
+ * - auth-conf: authentication + integrity + confidentiality (encryption)
+ */
+ private final String saslQop;
+
+ /**
+ * Accumulo service primary name for SASL authentication.
+ * Default is "accumulo".
+ */
+ private final String accumuloServicePrimary;
+
+ /**
+ * Whether to use delegation tokens for distributed execution.
+ * When true, the service will obtain delegation tokens for query users.
+ */
+ private final boolean useDelegationTokens;
+
+ // ===== Optional settings =====
+
+ /**
+ * Optional name of the schema metadata table.
+ * If set, the plugin will look for table schema definitions in this Accumulo table.
+ * Default is "_drill_schema".
+ */
+ private final String schemaMetadataTable;
+
+ /**
+ * Timeout in milliseconds for Accumulo client operations.
+ * Default is 30000 (30 seconds).
+ */
+ private final Integer clientTimeout;
+
+ /**
+ * Number of threads for BatchScanner operations.
+ * Default is 10.
+ */
+ private final Integer batchScannerThreads;
+
+ @JsonCreator
+ public AccumuloStoragePluginConfig(
+ @JsonProperty("zookeeperQuorum") String zookeeperQuorum,
+ @JsonProperty("instanceName") String instanceName,
+ @JsonProperty("username") String username,
+ @JsonProperty("password") String password,
+ @JsonProperty("authenticationType") String authenticationType,
+ @JsonProperty("principal") String principal,
+ @JsonProperty("keytabPath") String keytabPath,
+ @JsonProperty("saslQop") String saslQop,
+ @JsonProperty("accumuloServicePrimary") String accumuloServicePrimary,
+ @JsonProperty("useDelegationTokens") Boolean useDelegationTokens,
+ @JsonProperty("authMode") String authMode,
+ @JsonProperty("credentialsProvider") CredentialsProvider credentialsProvider,
+ @JsonProperty("schemaMetadataTable") String schemaMetadataTable,
+ @JsonProperty("clientTimeout") Integer clientTimeout,
+ @JsonProperty("batchScannerThreads") Integer batchScannerThreads) {
+
+ super(
+ CredentialProviderUtils.getCredentialsProvider(username, password, credentialsProvider),
+ credentialsProvider == null,
+ AuthMode.parseOrDefault(authMode, AuthMode.SHARED_USER)
+ );
+
+ this.zookeeperQuorum = zookeeperQuorum;
+ this.instanceName = instanceName;
+ this.username = username;
+ this.password = password;
+
+ this.authenticationType = AccumuloAuthType.parseOrDefault(authenticationType, AccumuloAuthType.PASSWORD);
+ this.principal = principal;
+ this.keytabPath = keytabPath;
+ this.saslQop = saslQop != null ? saslQop : DEFAULT_SASL_QOP;
+ this.accumuloServicePrimary = accumuloServicePrimary != null ? accumuloServicePrimary : DEFAULT_ACCUMULO_SERVICE_PRIMARY;
+ this.useDelegationTokens = useDelegationTokens != null ? useDelegationTokens : false;
+
+ this.schemaMetadataTable = schemaMetadataTable != null ? schemaMetadataTable : "_drill_schema";
+ this.clientTimeout = clientTimeout != null ? clientTimeout : 30000;
+ this.batchScannerThreads = batchScannerThreads != null ? batchScannerThreads : 10;
+ }
+
+ /**
+ * Simplified constructor for password authentication (backward compatible).
+ */
+ public AccumuloStoragePluginConfig(
+ String zookeeperQuorum,
+ String instanceName,
+ String username,
+ String password) {
+ this(zookeeperQuorum, instanceName, username, password,
+ null, null, null, null, null, null, null, null, null, null, null);
+ }
+
+ // ===== Connection Getters =====
+
+ @JsonProperty("zookeeperQuorum")
+ public String getZookeeperQuorum() {
+ return zookeeperQuorum;
+ }
+
+ @JsonProperty("instanceName")
+ public String getInstanceName() {
+ return instanceName;
+ }
+
+ // ===== Password Auth Getters =====
+
+ @JsonProperty("username")
+ public String getUsername() {
+ return username;
+ }
+
+ @JsonProperty("password")
+ public String getPassword() {
+ return password;
+ }
+
+ // ===== Kerberos Auth Getters =====
+
+ @JsonProperty("authenticationType")
+ public AccumuloAuthType getAuthenticationType() {
+ return authenticationType;
+ }
+
+ @JsonProperty("principal")
+ public String getPrincipal() {
+ return principal;
+ }
+
+ @JsonProperty("keytabPath")
+ public String getKeytabPath() {
+ return keytabPath;
+ }
+
+ @JsonProperty("saslQop")
+ public String getSaslQop() {
+ return saslQop;
+ }
+
+ @JsonProperty("accumuloServicePrimary")
+ public String getAccumuloServicePrimary() {
+ return accumuloServicePrimary;
+ }
+
+ @JsonProperty("useDelegationTokens")
+ public boolean isUseDelegationTokens() {
+ return useDelegationTokens;
+ }
+
+ // ===== Optional Settings Getters =====
+
+ @JsonProperty("schemaMetadataTable")
+ public String getSchemaMetadataTable() {
+ return schemaMetadataTable;
+ }
+
+ @JsonProperty("clientTimeout")
+ public Integer getClientTimeout() {
+ return clientTimeout;
+ }
+
+ @JsonProperty("batchScannerThreads")
+ public Integer getBatchScannerThreads() {
+ return batchScannerThreads;
+ }
+
+ // ===== Credential Helper Methods =====
+
+ /**
+ * Returns username/password credentials for the specified user context.
+ *
+ * For SHARED_USER mode, returns the configured credentials. + * For USER_TRANSLATION mode, returns per-user credentials from the provider.
+ * + * @param userCredentials the query user credentials (may be null for SHARED_USER) + * @return Optional containing credentials if available + */ + @JsonIgnore + public OptionalIn the future, this will represent a scan on specific tablets. + * For now, it represents a full table scan.
+ * + *For user impersonation mode, this class carries a delegation token + * that was generated during planning and is used at execution time to + * create an Accumulo client with the user's identity.
+ */ +@JsonTypeName("accumulo-sub-scan") +public class AccumuloSubScan extends AbstractBase implements SubScan { + + public static final String OPERATOR_TYPE = "ACCUMULO_SUB_SCAN"; + + private final AccumuloStoragePlugin storagePlugin; + private final AccumuloScanSpec scanSpec; + private final ListAccumulo stores all data as byte arrays. This class provides methods to + * convert those byte arrays to appropriate Java types based on the configured + * column type in the schema.
+ * + *Conversion strategies:
+ *This class enables delegation tokens to be passed across Drill's distributed + * execution pipeline via JSON serialization. The token is stored as a Base64-encoded + * string for safe transport through JSON.
+ * + *Usage flow:
+ *Note: This returns an AuthenticationToken (the parent interface) rather than + * DelegationToken because the deserialization uses the stored class name.
+ * + * @return the deserialized AuthenticationToken + */ + @SuppressWarnings("unchecked") + @JsonIgnore + public AuthenticationToken toAuthenticationToken() { + byte[] tokenBytes = Base64.getDecoder().decode(serializedToken); + try { + Class extends AuthenticationToken> tokenClass = + (Class extends AuthenticationToken>) Class.forName(tokenClassName); + return AuthenticationTokenSerializer.deserialize(tokenClass, tokenBytes); + } catch (ClassNotFoundException e) { + throw new RuntimeException("Failed to load token class: " + tokenClassName, e); + } + } + + @JsonProperty("userName") + public String getUserName() { + return userName; + } + + @JsonProperty("serializedToken") + public String getSerializedToken() { + return serializedToken; + } + + @JsonProperty("tokenClassName") + public String getTokenClassName() { + return tokenClassName; + } + + @JsonProperty("creationTime") + public long getCreationTime() { + return creationTime; + } + + /** + * Returns the age of this token in milliseconds. + */ + @JsonIgnore + public long getAgeMillis() { + return System.currentTimeMillis() - creationTime; + } + + /** + * Checks if this token is older than the specified age. + * + * @param maxAgeMillis maximum acceptable age in milliseconds + * @return true if the token is older than maxAgeMillis + */ + @JsonIgnore + public boolean isOlderThan(long maxAgeMillis) { + return getAgeMillis() > maxAgeMillis; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + DelegationTokenInfo that = (DelegationTokenInfo) o; + return creationTime == that.creationTime + && Objects.equals(userName, that.userName) + && Objects.equals(serializedToken, that.serializedToken) + && Objects.equals(tokenClassName, that.tokenClassName); + } + + @Override + public int hashCode() { + return Objects.hash(userName, serializedToken, tokenClassName, creationTime); + } + + @Override + public String toString() { + return new PlanStringBuilder(this) + .field("userName", userName) + .field("tokenClassName", tokenClassName) + .field("creationTime", creationTime) + .field("tokenLength", serializedToken != null ? serializedToken.length() : 0) + .toString(); + } +} diff --git a/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/DrillAccumuloConstants.java b/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/DrillAccumuloConstants.java new file mode 100644 index 00000000000..bfde8b6a2b5 --- /dev/null +++ b/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/DrillAccumuloConstants.java @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo; + +import org.apache.drill.common.expression.SchemaPath; +import org.apache.drill.common.types.TypeProtos.MajorType; +import org.apache.drill.common.types.TypeProtos.MinorType; +import org.apache.drill.common.types.Types; + +/** + * Constants used by the Accumulo storage plugin. + */ +public interface DrillAccumuloConstants { + + /** + * Name of the row key column in Drill queries. + */ + String ROW_KEY = "row_key"; + + /** + * Schema path for the row key. + */ + SchemaPath ROW_KEY_PATH = SchemaPath.getSimplePath(ROW_KEY); + + /** + * Type for the row key column (required VARBINARY). + */ + MajorType ROW_KEY_TYPE = Types.required(MinorType.VARBINARY); + + /** + * Type for column family maps (required MAP). + */ + MajorType COLUMN_FAMILY_TYPE = Types.required(MinorType.MAP); + + /** + * Type for individual columns within a family (optional VARBINARY). + */ + MajorType COLUMN_TYPE = Types.optional(MinorType.VARBINARY); + + /** + * Separator between column family and qualifier in Accumulo keys. + */ + String COLUMN_SEPARATOR = ":"; + + /** + * Default batch size for scanner caching. + */ + int DEFAULT_BATCH_SIZE = 4000; +} diff --git a/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/DrillAccumuloTable.java b/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/DrillAccumuloTable.java new file mode 100644 index 00000000000..5f136d512fa --- /dev/null +++ b/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/DrillAccumuloTable.java @@ -0,0 +1,190 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo; + +import java.util.ArrayList; +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Set; + +import org.apache.accumulo.core.client.Scanner; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.security.Authorizations; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.sql.type.SqlTypeName; +import org.apache.drill.exec.planner.logical.DrillTable; +import org.apache.drill.exec.store.accumulo.schema.ColumnDef; +import org.apache.drill.exec.store.accumulo.schema.TableSchema; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Represents an Accumulo table in Drill's query planner. + * + *This class provides the row type (schema) for Accumulo tables to Drill's + * Calcite-based query planner. It uses the configured schema provider to + * discover column definitions.
+ * + *Schema resolution follows this order:
+ *Unlike HBase, Accumulo does not declare its column families up front, so they + * are read from the first {@value #COLUMN_FAMILY_SAMPLE_SIZE} entries. A family that + * appears only beyond that point will not be visible to the planner; define an + * explicit schema in the metadata table for tables where that matters.
+ */ + private SetAccumulo is schema-less and stores all data as byte arrays. + * This enum defines the logical types that Drill will use to interpret + * the byte data when reading from Accumulo.
+ */ +public enum AccumuloColumnType { + + /** + * Variable-length string (default type). + */ + VARCHAR(SqlTypeName.VARCHAR), + + /** + * Fixed-length string. + */ + CHAR(SqlTypeName.CHAR), + + /** + * 32-bit signed integer. + */ + INT(SqlTypeName.INTEGER), + + /** + * Alias for INT. + */ + INTEGER(SqlTypeName.INTEGER), + + /** + * 64-bit signed integer. + */ + BIGINT(SqlTypeName.BIGINT), + + /** + * Alias for BIGINT. + */ + LONG(SqlTypeName.BIGINT), + + /** + * 16-bit signed integer. + */ + SMALLINT(SqlTypeName.SMALLINT), + + /** + * 8-bit signed integer. + */ + TINYINT(SqlTypeName.TINYINT), + + /** + * Single-precision floating point. + */ + FLOAT(SqlTypeName.FLOAT), + + /** + * Double-precision floating point. + */ + DOUBLE(SqlTypeName.DOUBLE), + + /** + * Exact numeric with configurable precision and scale. + */ + DECIMAL(SqlTypeName.DECIMAL), + + /** + * Boolean value. + */ + BOOLEAN(SqlTypeName.BOOLEAN), + + /** + * Date without time component. + */ + DATE(SqlTypeName.DATE), + + /** + * Time without date component. + */ + TIME(SqlTypeName.TIME), + + /** + * Date and time. + */ + TIMESTAMP(SqlTypeName.TIMESTAMP), + + /** + * Binary data (raw bytes). + */ + VARBINARY(SqlTypeName.VARBINARY), + + /** + * Dynamic type (determined at runtime). + */ + ANY(SqlTypeName.ANY); + + private final SqlTypeName sqlTypeName; + + AccumuloColumnType(SqlTypeName sqlTypeName) { + this.sqlTypeName = sqlTypeName; + } + + /** + * Returns the corresponding Calcite SQL type name. + */ + public SqlTypeName getSqlTypeName() { + return sqlTypeName; + } + + /** + * Parses a type string to an AccumuloColumnType. + * + *Case-insensitive matching. Returns VARCHAR if the type is not recognized.
+ * + * @param typeString the type string to parse + * @return the corresponding AccumuloColumnType + */ + public static AccumuloColumnType fromString(String typeString) { + if (typeString == null || typeString.trim().isEmpty()) { + return VARCHAR; + } + + String normalized = typeString.trim().toUpperCase(); + + // Handle common aliases + switch (normalized) { + case "STRING": + case "TEXT": + return VARCHAR; + case "INT": + case "INTEGER": + return INTEGER; + case "LONG": + case "BIGINT": + return BIGINT; + case "FLOAT": + case "REAL": + return FLOAT; + case "DOUBLE": + case "DOUBLE PRECISION": + return DOUBLE; + case "BOOL": + case "BOOLEAN": + return BOOLEAN; + case "BYTES": + case "BINARY": + case "VARBINARY": + return VARBINARY; + default: + try { + return valueOf(normalized); + } catch (IllegalArgumentException e) { + return VARCHAR; + } + } + } + + /** + * Returns true if this type is a numeric type. + */ + public boolean isNumeric() { + switch (this) { + case INT: + case INTEGER: + case BIGINT: + case LONG: + case SMALLINT: + case TINYINT: + case FLOAT: + case DOUBLE: + case DECIMAL: + return true; + default: + return false; + } + } + + /** + * Returns true if this type is a temporal type. + */ + public boolean isTemporal() { + switch (this) { + case DATE: + case TIME: + case TIMESTAMP: + return true; + default: + return false; + } + } +} diff --git a/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/schema/AccumuloSchemaProvider.java b/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/schema/AccumuloSchemaProvider.java new file mode 100644 index 00000000000..d775723eaf5 --- /dev/null +++ b/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/schema/AccumuloSchemaProvider.java @@ -0,0 +1,89 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo.schema; + +import java.util.Set; + +import org.apache.accumulo.core.client.AccumuloClient; + +/** + * Interface for schema discovery strategies in the Accumulo storage plugin. + * + *This interface abstracts how table schemas are discovered from Accumulo. + * Different implementations can provide schema information from different sources:
+ *This is the extension point for Option B (advanced mode) where custom schema + * providers could expose Accumulo-specific features like iterators.
+ */ +public interface AccumuloSchemaProvider { + + /** + * Returns the schema for the specified Accumulo table. + * + *If the schema is not found or cannot be determined, implementations should + * return a dynamic schema (via {@link TableSchema#dynamic(String)}) rather than + * throwing an exception.
+ * + * @param client the Accumulo client + * @param tableName the name of the table + * @return the table schema, never null + */ + TableSchema getTableSchema(AccumuloClient client, String tableName); + + /** + * Discovers all table names available in the Accumulo instance. + * + *Implementations should filter out system tables (e.g., tables starting with "accumulo.") + * unless specifically configured to include them.
+ * + * @param client the Accumulo client + * @return set of table names, never null (may be empty) + */ + SetThis can be used to check if explicit schema metadata exists before + * falling back to dynamic schema discovery.
+ * + * @param client the Accumulo client + * @param tableName the name of the table + * @return true if schema information is available + */ + boolean hasSchema(AccumuloClient client, String tableName); + + /** + * Clears any cached schema information. + * + *Called when schema metadata may have changed and needs to be refreshed.
+ */ + void clearCache(); + + /** + * Clears cached schema information for a specific table. + * + * @param tableName the table to clear from cache + */ + void clearCache(String tableName); +} diff --git a/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/schema/ColumnDef.java b/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/schema/ColumnDef.java new file mode 100644 index 00000000000..ab6b1c238d6 --- /dev/null +++ b/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/schema/ColumnDef.java @@ -0,0 +1,146 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo.schema; + +import java.util.Objects; + +import org.apache.calcite.sql.type.SqlTypeName; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** + * Represents a column definition for an Accumulo table in Drill. + * + *Maps an Accumulo column family:qualifier pair to a Drill column with + * a specific SQL type.
+ */ +public class ColumnDef { + + private final String name; + private final String columnFamily; + private final String columnQualifier; + private final AccumuloColumnType type; + private final boolean nullable; + + @JsonCreator + public ColumnDef( + @JsonProperty("name") String name, + @JsonProperty("columnFamily") String columnFamily, + @JsonProperty("columnQualifier") String columnQualifier, + @JsonProperty("type") AccumuloColumnType type, + @JsonProperty("nullable") Boolean nullable) { + this.name = name; + this.columnFamily = columnFamily; + this.columnQualifier = columnQualifier; + this.type = type != null ? type : AccumuloColumnType.VARCHAR; + this.nullable = nullable != null ? nullable : true; + } + + /** + * Convenience constructor for creating a column definition. + */ + public static ColumnDef create(String name, String columnFamily, String columnQualifier, + AccumuloColumnType type) { + return new ColumnDef(name, columnFamily, columnQualifier, type, true); + } + + /** + * Convenience constructor for VARCHAR columns. + */ + public static ColumnDef varchar(String name, String columnFamily, String columnQualifier) { + return new ColumnDef(name, columnFamily, columnQualifier, AccumuloColumnType.VARCHAR, true); + } + + @JsonProperty("name") + public String getName() { + return name; + } + + @JsonProperty("columnFamily") + public String getColumnFamily() { + return columnFamily; + } + + @JsonProperty("columnQualifier") + public String getColumnQualifier() { + return columnQualifier; + } + + @JsonProperty("type") + public AccumuloColumnType getType() { + return type; + } + + @JsonProperty("nullable") + public boolean isNullable() { + return nullable; + } + + /** + * Returns the SQL type name for this column. + */ + @JsonIgnore + public SqlTypeName getSqlTypeName() { + return type.getSqlTypeName(); + } + + /** + * Returns the full Accumulo column identifier (family:qualifier). + */ + @JsonIgnore + public String getFullColumnName() { + if (columnQualifier == null || columnQualifier.isEmpty()) { + return columnFamily; + } + return columnFamily + ":" + columnQualifier; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + ColumnDef columnDef = (ColumnDef) o; + return nullable == columnDef.nullable + && Objects.equals(name, columnDef.name) + && Objects.equals(columnFamily, columnDef.columnFamily) + && Objects.equals(columnQualifier, columnDef.columnQualifier) + && type == columnDef.type; + } + + @Override + public int hashCode() { + return Objects.hash(name, columnFamily, columnQualifier, type, nullable); + } + + @Override + public String toString() { + return "ColumnDef{" + + "name='" + name + '\'' + + ", columnFamily='" + columnFamily + '\'' + + ", columnQualifier='" + columnQualifier + '\'' + + ", type=" + type + + ", nullable=" + nullable + + '}'; + } +} diff --git a/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/schema/MetadataTableSchemaProvider.java b/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/schema/MetadataTableSchemaProvider.java new file mode 100644 index 00000000000..b6b555f7d19 --- /dev/null +++ b/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/schema/MetadataTableSchemaProvider.java @@ -0,0 +1,293 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo.schema; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; + +import org.apache.accumulo.core.client.AccumuloClient; +import org.apache.accumulo.core.client.Scanner; +import org.apache.accumulo.core.client.TableNotFoundException; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Range; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.security.Authorizations; +import org.apache.hadoop.io.Text; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; + +/** + * Schema provider that reads table schema from an Accumulo metadata table. + * + *The metadata table stores schema information in the following format:
+ *
+ * Row Key: {table_name}
+ * Column Family: "schema"
+ * Column Qualifiers:
+ * - "row_key_type": The type of the row key (e.g., "VARCHAR", "VARBINARY")
+ * - "columns": JSON array of column definitions
+ *
+ * Example:
+ * Row: "users"
+ * schema:row_key_type = "VARCHAR"
+ * schema:columns = [
+ * {"name":"name","columnFamily":"cf1","columnQualifier":"name","type":"VARCHAR","nullable":true},
+ * {"name":"age","columnFamily":"cf1","columnQualifier":"age","type":"INT","nullable":true}
+ * ]
+ *
+ *
+ * This provider includes a configurable cache to reduce Accumulo metadata table lookups.
+ */ +public class MetadataTableSchemaProvider implements AccumuloSchemaProvider { + private static final Logger logger = LoggerFactory.getLogger(MetadataTableSchemaProvider.class); + + private static final String SCHEMA_COLUMN_FAMILY = "schema"; + private static final String ROW_KEY_TYPE_QUALIFIER = "row_key_type"; + private static final String COLUMNS_QUALIFIER = "columns"; + + private static final long DEFAULT_CACHE_TTL_MS = TimeUnit.MINUTES.toMillis(5); + + private final String metadataTableName; + private final ObjectMapper objectMapper; + private final MapThis is a utility method for setting up schema metadata. + * It creates the metadata table if it doesn't exist.
+ * + * @param client the Accumulo client + * @param schema the table schema to write + * @throws Exception if the write fails + */ + public void writeSchema(AccumuloClient client, TableSchema schema) throws Exception { + // Create metadata table if it doesn't exist + if (!client.tableOperations().exists(metadataTableName)) { + client.tableOperations().create(metadataTableName); + logger.info("Created metadata table: {}", metadataTableName); + } + + // Write schema to metadata table + try (var writer = client.createBatchWriter(metadataTableName)) { + org.apache.accumulo.core.data.Mutation mutation = + new org.apache.accumulo.core.data.Mutation(schema.getTableName()); + + // Write row key type + mutation.put(SCHEMA_COLUMN_FAMILY, ROW_KEY_TYPE_QUALIFIER, + schema.getRowKeyType().name()); + + // Write columns as JSON + String columnsJson = objectMapper.writeValueAsString(schema.getColumns()); + mutation.put(SCHEMA_COLUMN_FAMILY, COLUMNS_QUALIFIER, columnsJson); + + writer.addMutation(mutation); + } + + // Clear cache for this table + clearCache(schema.getTableName()); + logger.info("Wrote schema for table: {}", schema.getTableName()); + } +} diff --git a/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/schema/TableSchema.java b/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/schema/TableSchema.java new file mode 100644 index 00000000000..a2503b4f867 --- /dev/null +++ b/contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/schema/TableSchema.java @@ -0,0 +1,193 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo.schema; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** + * Represents the schema of an Accumulo table for Drill. + * + *Contains the table name, row key type, and column definitions. + * The schema is used by Drill's query planner to understand the structure + * of Accumulo tables.
+ */ +public class TableSchema { + + public static final String ROW_KEY_COLUMN = "row_key"; + + private final String tableName; + private final AccumuloColumnType rowKeyType; + private final ListThese tests verify basic SELECT queries work correctly against + * real Accumulo tables via MiniAccumuloCluster.
+ */ +public class AccumuloBasicQueryTest extends BaseAccumuloTest { + + @Test + public void testSelectStarFromTable1() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t"; + runAccumuloSQLVerifyCount(sql, 10); + } + + @Test + public void testSelectSpecificColumnsFromTable1() throws Exception { + String sql = "SELECT row_key, t.cf.name, t.cf.age FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t"; + runAccumuloSQLVerifyCount(sql, 10); + } + + @Test + public void testSelectRowKeyOnly() throws Exception { + String sql = "SELECT row_key FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t"; + runAccumuloSQLVerifyCount(sql, 10); + } + + @Test + public void testSelectFromUsersTable() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_USERS) + " t"; + runAccumuloSQLVerifyCount(sql, 20); + } + + @Test + public void testSelectMultipleColumnFamilies() throws Exception { + String sql = "SELECT row_key, t.personal.first_name, t.personal.last_name, t.employment.company " + + "FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_USERS) + " t"; + runAccumuloSQLVerifyCount(sql, 20); + } + + @Test + public void testSelectFromLargeTable() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_LARGE) + " t"; + runAccumuloSQLVerifyCount(sql, 1000); + } + + @Test + public void testCountStar() throws Exception { + String sql = "SELECT COUNT(*) FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t"; + runAccumuloSQLVerifyCount(sql, 1); + } + + @Test + public void testCountStarUsersTable() throws Exception { + String sql = "SELECT COUNT(*) FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_USERS) + " t"; + runAccumuloSQLVerifyCount(sql, 1); + } + + @Test + public void testDistinctCompany() throws Exception { + String sql = "SELECT DISTINCT t.employment.company FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_USERS) + " t"; + // Should have 3 distinct companies: Acme Corp, TechCo, DataInc + runAccumuloSQLVerifyCount(sql, 3); + } + + @Test + public void testGroupByCompany() throws Exception { + String sql = "SELECT t.employment.company, COUNT(*) as cnt " + + "FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_USERS) + " t" + + " GROUP BY t.employment.company"; + runAccumuloSQLVerifyCount(sql, 3); + } +} diff --git a/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloFilterBuilderTest.java b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloFilterBuilderTest.java new file mode 100644 index 00000000000..515c98ad481 --- /dev/null +++ b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloFilterBuilderTest.java @@ -0,0 +1,207 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.nio.charset.StandardCharsets; + +import org.apache.drill.common.FunctionNames; +import org.apache.drill.common.expression.FunctionCall; +import org.apache.drill.common.expression.SchemaPath; +import org.apache.drill.common.expression.ValueExpressions; +import org.apache.drill.test.BaseTest; +import org.junit.Test; + +import com.google.common.collect.ImmutableList; + +/** + * Unit tests for AccumuloCompareFunctionsProcessor and AccumuloFilterBuilder. + */ +public class AccumuloFilterBuilderTest extends BaseTest { + + @Test + public void testIsCompareFunction() { + assertTrue(AccumuloCompareFunctionsProcessor.isCompareFunction(FunctionNames.EQ)); + assertTrue(AccumuloCompareFunctionsProcessor.isCompareFunction(FunctionNames.NE)); + assertTrue(AccumuloCompareFunctionsProcessor.isCompareFunction(FunctionNames.LT)); + assertTrue(AccumuloCompareFunctionsProcessor.isCompareFunction(FunctionNames.LE)); + assertTrue(AccumuloCompareFunctionsProcessor.isCompareFunction(FunctionNames.GT)); + assertTrue(AccumuloCompareFunctionsProcessor.isCompareFunction(FunctionNames.GE)); + assertTrue(AccumuloCompareFunctionsProcessor.isCompareFunction(FunctionNames.IS_NULL)); + assertTrue(AccumuloCompareFunctionsProcessor.isCompareFunction(FunctionNames.IS_NOT_NULL)); + + assertFalse(AccumuloCompareFunctionsProcessor.isCompareFunction("unknown")); + assertFalse(AccumuloCompareFunctionsProcessor.isCompareFunction(FunctionNames.AND)); + assertFalse(AccumuloCompareFunctionsProcessor.isCompareFunction(FunctionNames.OR)); + } + + @Test + public void testProcessEqualFunction() { + // row_key = 'test' + SchemaPath path = SchemaPath.getSimplePath("row_key"); + ValueExpressions.QuotedString value = new ValueExpressions.QuotedString("test", 0, null); + + FunctionCall call = new FunctionCall( + FunctionNames.EQ, + ImmutableList.of(path, value), + null); + + AccumuloCompareFunctionsProcessor processor = + AccumuloCompareFunctionsProcessor.createFunctionsProcessorInstance(call); + + assertTrue(processor.isSuccess()); + assertEquals("row_key", processor.getPath().getRootSegmentPath()); + assertEquals("test", new String(processor.getValue(), StandardCharsets.UTF_8)); + assertEquals(FunctionNames.EQ, processor.getFunctionName()); + } + + @Test + public void testProcessGreaterThanFunction() { + // row_key > 'start' + SchemaPath path = SchemaPath.getSimplePath("row_key"); + ValueExpressions.QuotedString value = new ValueExpressions.QuotedString("start", 0, null); + + FunctionCall call = new FunctionCall( + FunctionNames.GT, + ImmutableList.of(path, value), + null); + + AccumuloCompareFunctionsProcessor processor = + AccumuloCompareFunctionsProcessor.createFunctionsProcessorInstance(call); + + assertTrue(processor.isSuccess()); + assertEquals("row_key", processor.getPath().getRootSegmentPath()); + assertEquals("start", new String(processor.getValue(), StandardCharsets.UTF_8)); + assertEquals(FunctionNames.GT, processor.getFunctionName()); + } + + @Test + public void testProcessLessThanFunction() { + // row_key < 'end' + SchemaPath path = SchemaPath.getSimplePath("row_key"); + ValueExpressions.QuotedString value = new ValueExpressions.QuotedString("end", 0, null); + + FunctionCall call = new FunctionCall( + FunctionNames.LT, + ImmutableList.of(path, value), + null); + + AccumuloCompareFunctionsProcessor processor = + AccumuloCompareFunctionsProcessor.createFunctionsProcessorInstance(call); + + assertTrue(processor.isSuccess()); + assertEquals(FunctionNames.LT, processor.getFunctionName()); + } + + @Test + public void testProcessSwappedOperands() { + // 'test' = row_key (value on left) + SchemaPath path = SchemaPath.getSimplePath("row_key"); + ValueExpressions.QuotedString value = new ValueExpressions.QuotedString("test", 0, null); + + FunctionCall call = new FunctionCall( + FunctionNames.EQ, + ImmutableList.of(value, path), + null); + + AccumuloCompareFunctionsProcessor processor = + AccumuloCompareFunctionsProcessor.createFunctionsProcessorInstance(call); + + assertTrue(processor.isSuccess()); + assertEquals("row_key", processor.getPath().getRootSegmentPath()); + assertEquals("test", new String(processor.getValue(), StandardCharsets.UTF_8)); + // Function should remain EQ since it's symmetric + assertEquals(FunctionNames.EQ, processor.getFunctionName()); + } + + @Test + public void testProcessSwappedGreaterThan() { + // 'value' > row_key should become row_key < 'value' + SchemaPath path = SchemaPath.getSimplePath("row_key"); + ValueExpressions.QuotedString value = new ValueExpressions.QuotedString("value", 0, null); + + FunctionCall call = new FunctionCall( + FunctionNames.GT, + ImmutableList.of(value, path), + null); + + AccumuloCompareFunctionsProcessor processor = + AccumuloCompareFunctionsProcessor.createFunctionsProcessorInstance(call); + + assertTrue(processor.isSuccess()); + assertEquals("row_key", processor.getPath().getRootSegmentPath()); + // GT transposes to LT when operands are swapped + assertEquals(FunctionNames.LT, processor.getFunctionName()); + } + + @Test + public void testProcessIntegerValue() { + // row_key = 123 + SchemaPath path = SchemaPath.getSimplePath("row_key"); + ValueExpressions.IntExpression value = new ValueExpressions.IntExpression(123, null); + + FunctionCall call = new FunctionCall( + FunctionNames.EQ, + ImmutableList.of(path, value), + null); + + AccumuloCompareFunctionsProcessor processor = + AccumuloCompareFunctionsProcessor.createFunctionsProcessorInstance(call); + + assertTrue(processor.isSuccess()); + assertEquals("123", new String(processor.getValue(), StandardCharsets.UTF_8)); + } + + @Test + public void testProcessLongValue() { + // row_key = 9999999999 + SchemaPath path = SchemaPath.getSimplePath("row_key"); + ValueExpressions.LongExpression value = new ValueExpressions.LongExpression(9999999999L, null); + + FunctionCall call = new FunctionCall( + FunctionNames.EQ, + ImmutableList.of(path, value), + null); + + AccumuloCompareFunctionsProcessor processor = + AccumuloCompareFunctionsProcessor.createFunctionsProcessorInstance(call); + + assertTrue(processor.isSuccess()); + assertEquals("9999999999", new String(processor.getValue(), StandardCharsets.UTF_8)); + } + + @Test + public void testCompareTransposeMap() { + // Verify the transpose map entries + assertEquals(FunctionNames.LE, + AccumuloCompareFunctionsProcessor.COMPARE_FUNCTIONS_TRANSPOSE_MAP.get(FunctionNames.GE)); + assertEquals(FunctionNames.LT, + AccumuloCompareFunctionsProcessor.COMPARE_FUNCTIONS_TRANSPOSE_MAP.get(FunctionNames.GT)); + assertEquals(FunctionNames.GE, + AccumuloCompareFunctionsProcessor.COMPARE_FUNCTIONS_TRANSPOSE_MAP.get(FunctionNames.LE)); + assertEquals(FunctionNames.GT, + AccumuloCompareFunctionsProcessor.COMPARE_FUNCTIONS_TRANSPOSE_MAP.get(FunctionNames.LT)); + assertEquals(FunctionNames.EQ, + AccumuloCompareFunctionsProcessor.COMPARE_FUNCTIONS_TRANSPOSE_MAP.get(FunctionNames.EQ)); + assertEquals(FunctionNames.NE, + AccumuloCompareFunctionsProcessor.COMPARE_FUNCTIONS_TRANSPOSE_MAP.get(FunctionNames.NE)); + } +} diff --git a/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloIntegrationTestsSuite.java b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloIntegrationTestsSuite.java new file mode 100644 index 00000000000..b50ea4141b7 --- /dev/null +++ b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloIntegrationTestsSuite.java @@ -0,0 +1,235 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo; + +import java.io.File; +import java.io.IOException; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.accumulo.core.client.AccumuloClient; +import org.apache.accumulo.core.client.security.tokens.PasswordToken; +import org.apache.accumulo.minicluster.MiniAccumuloCluster; +import org.apache.accumulo.minicluster.MiniAccumuloConfig; +import org.apache.drill.test.BaseTest; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.runner.RunWith; +import org.junit.runners.Suite; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Integration test suite for Accumulo storage plugin. + * + *This suite manages the lifecycle of a MiniAccumuloCluster and runs + * all integration tests that require a real Accumulo instance.
+ * + *Run with: {@code mvn test -Dtest=AccumuloIntegrationTestsSuite}
+ */ +@RunWith(Suite.class) +@Suite.SuiteClasses({ + AccumuloBasicQueryTest.class, + AccumuloPushdownIntegrationTest.class, + AccumuloResultVerificationTest.class, + AccumuloSerDeTest.class +}) +public class AccumuloIntegrationTestsSuite extends BaseTest { + private static final Logger logger = LoggerFactory.getLogger(AccumuloIntegrationTestsSuite.class); + + public static final String ROOT_USER = "root"; + public static final String ROOT_PASSWORD = "drilltest"; + public static final String INSTANCE_NAME = "drill-accumulo-test"; + + private static MiniAccumuloCluster miniCluster; + private static AccumuloClient client; + private static File tempDir; + private static volatile AtomicInteger initCount = new AtomicInteger(0); + private static boolean clusterStarted = false; + private static boolean tablesCreated = false; + + /** + * Whether to manage the MiniAccumuloCluster (start/stop). + * Set to false to use an external Accumulo instance. + */ + private static boolean manageMiniCluster = Boolean.parseBoolean( + System.getProperty("drill.accumulo.tests.managed", "true")); + + /** + * Whether to create test tables. + */ + private static boolean createTables = Boolean.parseBoolean( + System.getProperty("drill.accumulo.tests.createTables", "true")); + + @BeforeClass + public static void initCluster() throws Exception { + if (initCount.get() == 0) { + synchronized (AccumuloIntegrationTestsSuite.class) { + if (initCount.get() == 0) { + if (manageMiniCluster) { + startMiniCluster(); + } else { + connectToExternalCluster(); + } + + if (createTables) { + createTestTables(); + } + + initCount.incrementAndGet(); + return; + } + } + } + initCount.incrementAndGet(); + } + + @AfterClass + public static void tearDownCluster() throws Exception { + synchronized (AccumuloIntegrationTestsSuite.class) { + if (initCount.decrementAndGet() == 0) { + if (createTables && tablesCreated) { + cleanupTestTables(); + } + + if (client != null) { + client.close(); + client = null; + } + + if (clusterStarted && miniCluster != null) { + logger.info("Stopping MiniAccumuloCluster..."); + miniCluster.stop(); + miniCluster = null; + logger.info("MiniAccumuloCluster stopped."); + } + + // Clean up temp directory + if (tempDir != null && tempDir.exists()) { + deleteDirectory(tempDir); + } + } + } + } + + private static void startMiniCluster() throws Exception { + logger.info("Starting MiniAccumuloCluster..."); + + // Create temp directory for cluster data + tempDir = new File(System.getProperty("accumulo.test.root", + System.getProperty("java.io.tmpdir")), "mini-accumulo-" + System.currentTimeMillis()); + if (!tempDir.mkdirs()) { + throw new IOException("Failed to create temp directory: " + tempDir); + } + + MiniAccumuloConfig config = new MiniAccumuloConfig(tempDir, ROOT_PASSWORD); + config.setInstanceName(INSTANCE_NAME); + config.setNumTservers(1); + + miniCluster = new MiniAccumuloCluster(config); + miniCluster.start(); + clusterStarted = true; + + // Create client + client = miniCluster.createAccumuloClient(ROOT_USER, new PasswordToken(ROOT_PASSWORD)); + + logger.info("MiniAccumuloCluster started. Instance: {}, ZooKeepers: {}", + miniCluster.getInstanceName(), miniCluster.getZooKeepers()); + } + + private static void connectToExternalCluster() throws Exception { + String zookeepers = System.getProperty("drill.accumulo.zookeepers", "localhost:2181"); + String instanceName = System.getProperty("drill.accumulo.instance", "accumulo"); + String user = System.getProperty("drill.accumulo.user", "root"); + String password = System.getProperty("drill.accumulo.password", "secret"); + + logger.info("Connecting to external Accumulo instance: {} at {}", instanceName, zookeepers); + + client = org.apache.accumulo.core.client.Accumulo.newClient() + .to(instanceName, zookeepers) + .as(user, password) + .build(); + } + + private static void createTestTables() throws Exception { + logger.info("Creating test tables..."); + AccumuloTestUtils.createAllTestTables(client); + tablesCreated = true; + logger.info("Test tables created."); + } + + private static void cleanupTestTables() { + try { + logger.info("Cleaning up test tables..."); + AccumuloTestUtils.deleteAllTestTables(client); + logger.info("Test tables cleaned up."); + } catch (Exception e) { + logger.warn("Error cleaning up test tables", e); + } + } + + private static void deleteDirectory(File dir) { + File[] files = dir.listFiles(); + if (files != null) { + for (File file : files) { + if (file.isDirectory()) { + deleteDirectory(file); + } else { + file.delete(); + } + } + } + dir.delete(); + } + + // Public accessors for test classes + + public static MiniAccumuloCluster getMiniCluster() { + return miniCluster; + } + + public static AccumuloClient getClient() { + return client; + } + + public static String getZooKeepers() { + if (miniCluster != null) { + return miniCluster.getZooKeepers(); + } + return System.getProperty("drill.accumulo.zookeepers", "localhost:2181"); + } + + public static String getInstanceName() { + if (miniCluster != null) { + return miniCluster.getInstanceName(); + } + return System.getProperty("drill.accumulo.instance", "accumulo"); + } + + public static String getRootUser() { + return ROOT_USER; + } + + public static String getRootPassword() { + return ROOT_PASSWORD; + } + + public static void configure(boolean manageMiniCluster, boolean createTables) { + AccumuloIntegrationTestsSuite.manageMiniCluster = manageMiniCluster; + AccumuloIntegrationTestsSuite.createTables = createTables; + } +} diff --git a/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloKerberosConfigTest.java b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloKerberosConfigTest.java new file mode 100644 index 00000000000..f0178f9bd1d --- /dev/null +++ b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloKerberosConfigTest.java @@ -0,0 +1,263 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import org.apache.drill.common.logical.StoragePluginConfig.AuthMode; +import org.apache.drill.test.BaseTest; +import org.junit.Test; + +import com.fasterxml.jackson.databind.ObjectMapper; + +/** + * Unit tests for Kerberos-specific configuration in AccumuloStoragePluginConfig. + */ +public class AccumuloKerberosConfigTest extends BaseTest { + + @Test + public void testAuthTypeEnum() { + assertEquals(AccumuloAuthType.PASSWORD, AccumuloAuthType.parseOrDefault(null, AccumuloAuthType.PASSWORD)); + assertEquals(AccumuloAuthType.PASSWORD, AccumuloAuthType.parseOrDefault("", AccumuloAuthType.PASSWORD)); + assertEquals(AccumuloAuthType.PASSWORD, AccumuloAuthType.parseOrDefault("PASSWORD", AccumuloAuthType.KERBEROS)); + assertEquals(AccumuloAuthType.KERBEROS, AccumuloAuthType.parseOrDefault("KERBEROS", AccumuloAuthType.PASSWORD)); + assertEquals(AccumuloAuthType.KERBEROS, AccumuloAuthType.parseOrDefault("kerberos", AccumuloAuthType.PASSWORD)); + assertEquals(AccumuloAuthType.KERBEROS, AccumuloAuthType.parseOrDefault("Kerberos", AccumuloAuthType.PASSWORD)); + } + + @Test(expected = IllegalArgumentException.class) + public void testAuthTypeEnumInvalidValue() { + AccumuloAuthType.parseOrDefault("INVALID", AccumuloAuthType.PASSWORD); + } + + @Test + public void testKerberosSharedUserConfig() { + AccumuloStoragePluginConfig config = new AccumuloStoragePluginConfig( + "zk:2181", + "accumulo", + null, + null, + "KERBEROS", + "drill/host@REALM", + "/etc/security/keytabs/drill.keytab", + "auth", + "accumulo", + false, + "SHARED_USER", + null, + null, + null, + null + ); + + assertTrue(config.isKerberosEnabled()); + assertFalse(config.isUserImpersonationEnabled()); + assertFalse(config.isUseDelegationTokens()); + assertEquals(AuthMode.SHARED_USER, config.getAuthMode()); + } + + @Test + public void testKerberosUserImpersonationConfig() { + AccumuloStoragePluginConfig config = new AccumuloStoragePluginConfig( + "zk:2181", + "accumulo", + null, + null, + "KERBEROS", + "drill/host@REALM", + "/etc/security/keytabs/drill.keytab", + "auth-conf", + "accumulo", + true, + "USER_IMPERSONATION", + null, + null, + null, + null + ); + + assertTrue(config.isKerberosEnabled()); + assertTrue(config.isUserImpersonationEnabled()); + assertTrue(config.isUseDelegationTokens()); + assertEquals(AuthMode.USER_IMPERSONATION, config.getAuthMode()); + assertEquals("auth-conf", config.getSaslQop()); + } + + @Test + public void testSaslQopValues() { + // Test auth + AccumuloStoragePluginConfig authConfig = new AccumuloStoragePluginConfig( + "zk:2181", "accumulo", null, null, + "KERBEROS", "drill@REALM", "/keytab", "auth", null, null, null, null, null, null, null + ); + assertEquals("auth", authConfig.getSaslQop()); + + // Test auth-int + AccumuloStoragePluginConfig authIntConfig = new AccumuloStoragePluginConfig( + "zk:2181", "accumulo", null, null, + "KERBEROS", "drill@REALM", "/keytab", "auth-int", null, null, null, null, null, null, null + ); + assertEquals("auth-int", authIntConfig.getSaslQop()); + + // Test auth-conf + AccumuloStoragePluginConfig authConfConfig = new AccumuloStoragePluginConfig( + "zk:2181", "accumulo", null, null, + "KERBEROS", "drill@REALM", "/keytab", "auth-conf", null, null, null, null, null, null, null + ); + assertEquals("auth-conf", authConfConfig.getSaslQop()); + } + + @Test + public void testBackwardCompatibilityPasswordAuth() { + // Old-style configuration without any Kerberos fields should still work + AccumuloStoragePluginConfig config = new AccumuloStoragePluginConfig( + "localhost:2181", + "accumulo", + "root", + "secret" + ); + + assertEquals(AccumuloAuthType.PASSWORD, config.getAuthenticationType()); + assertFalse(config.isKerberosEnabled()); + assertFalse(config.isUserImpersonationEnabled()); + assertEquals(AuthMode.SHARED_USER, config.getAuthMode()); + assertEquals("root", config.getUsername()); + assertEquals("secret", config.getPassword()); + } + + @Test + public void testJsonSerializationFullKerberosConfig() throws Exception { + ObjectMapper mapper = new ObjectMapper(); + + AccumuloStoragePluginConfig config = new AccumuloStoragePluginConfig( + "zk1:2181,zk2:2181", + "accumulo_prod", + null, + null, + "KERBEROS", + "drill/drillserver.example.com@EXAMPLE.COM", + "/etc/security/keytabs/drill.service.keytab", + "auth-conf", + "accumulo", + true, + "USER_IMPERSONATION", + null, + "_drill_schema", + 30000, + 10 + ); + + String json = mapper.writeValueAsString(config); + assertNotNull(json); + + // Verify JSON contains expected fields + assertTrue(json.contains("zk1:2181,zk2:2181")); + assertTrue(json.contains("accumulo_prod")); + assertTrue(json.contains("KERBEROS")); + assertTrue(json.contains("drill/drillserver.example.com@EXAMPLE.COM")); + assertTrue(json.contains("auth-conf")); + assertTrue(json.contains("\"useDelegationTokens\":true")); + + // Deserialize and verify + AccumuloStoragePluginConfig deserialized = mapper.readValue(json, AccumuloStoragePluginConfig.class); + assertEquals(config.getZookeeperQuorum(), deserialized.getZookeeperQuorum()); + assertEquals(config.getInstanceName(), deserialized.getInstanceName()); + assertEquals(config.getAuthenticationType(), deserialized.getAuthenticationType()); + assertEquals(config.getPrincipal(), deserialized.getPrincipal()); + assertEquals(config.getKeytabPath(), deserialized.getKeytabPath()); + assertEquals(config.getSaslQop(), deserialized.getSaslQop()); + assertEquals(config.getAccumuloServicePrimary(), deserialized.getAccumuloServicePrimary()); + assertEquals(config.isUseDelegationTokens(), deserialized.isUseDelegationTokens()); + } + + @Test + public void testMixedAuthConfigInvalid() { + // Config with both password creds and Kerberos settings + // This should be allowed for migration scenarios + AccumuloStoragePluginConfig config = new AccumuloStoragePluginConfig( + "zk:2181", + "accumulo", + "fallback_user", // password username + "fallback_pass", // password + "KERBEROS", // but auth type is Kerberos + "drill@REALM", + "/keytab", + null, + null, + null, + null, + null, + null, + null, + null + ); + + // When KERBEROS auth type is set, it should use Kerberos + assertEquals(AccumuloAuthType.KERBEROS, config.getAuthenticationType()); + assertTrue(config.isKerberosEnabled()); + // But password fields are still accessible if needed + assertEquals("fallback_user", config.getUsername()); + } + + @Test + public void testUserTranslationConfig() { + AccumuloStoragePluginConfig config = new AccumuloStoragePluginConfig( + "zk:2181", + "accumulo", + "service_user", // service account for fallback + "service_pass", + "PASSWORD", + null, + null, + null, + null, + false, + "USER_TRANSLATION", + null, + null, + null, + null + ); + + assertFalse(config.isKerberosEnabled()); + assertFalse(config.isUserImpersonationEnabled()); + assertTrue(config.isUserTranslationEnabled()); + assertEquals(AuthMode.USER_TRANSLATION, config.getAuthMode()); + } + + @Test + public void testPrincipalFormats() { + // Test simple principal (just user@REALM) + AccumuloStoragePluginConfig simpleConfig = new AccumuloStoragePluginConfig( + "zk:2181", "accumulo", null, null, + "KERBEROS", "drill@EXAMPLE.COM", "/keytab", null, null, null, null, null, null, null, null + ); + assertEquals("drill@EXAMPLE.COM", simpleConfig.getPrincipal()); + + // Test service principal (primary/instance@REALM) + AccumuloStoragePluginConfig serviceConfig = new AccumuloStoragePluginConfig( + "zk:2181", "accumulo", null, null, + "KERBEROS", "drill/drillserver.example.com@EXAMPLE.COM", "/keytab", + null, null, null, null, null, null, null, null + ); + assertEquals("drill/drillserver.example.com@EXAMPLE.COM", serviceConfig.getPrincipal()); + } +} diff --git a/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloLimitPushdownTest.java b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloLimitPushdownTest.java new file mode 100644 index 00000000000..b7c7d1744d9 --- /dev/null +++ b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloLimitPushdownTest.java @@ -0,0 +1,252 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.Arrays; +import java.util.List; + +import org.apache.drill.common.expression.SchemaPath; +import org.apache.drill.exec.physical.base.GroupScan; +import org.apache.drill.exec.physical.base.ScanStats; +import org.apache.drill.test.BaseTest; +import org.junit.Test; +import org.mockito.Mockito; + +/** + * Unit tests for limit pushdown in AccumuloGroupScan. + */ +public class AccumuloLimitPushdownTest extends BaseTest { + + /** + * Creates a mock AccumuloGroupScan for testing. + */ + private AccumuloGroupScan createTestGroupScan() { + AccumuloStoragePlugin mockPlugin = Mockito.mock(AccumuloStoragePlugin.class); + AccumuloStoragePluginConfig mockConfig = Mockito.mock(AccumuloStoragePluginConfig.class); + Mockito.when(mockPlugin.getConfig()).thenReturn(mockConfig); + + AccumuloScanSpec scanSpec = new AccumuloScanSpec("test_table"); + return new AccumuloGroupScan("testUser", mockPlugin, scanSpec, null, -1); + } + + @Test + public void testSupportsLimitPushdown() { + AccumuloGroupScan scan = createTestGroupScan(); + assertTrue("AccumuloGroupScan should support limit pushdown", scan.supportsLimitPushdown()); + } + + @Test + public void testApplyLimitReturnsNewScan() { + AccumuloGroupScan original = createTestGroupScan(); + + GroupScan newScan = original.applyLimit(100); + + assertNotNull("applyLimit should return a new scan", newScan); + assertNotSame("applyLimit should return a different instance", original, newScan); + assertTrue(newScan instanceof AccumuloGroupScan); + } + + @Test + public void testApplyLimitSetsMaxRecords() { + AccumuloGroupScan original = createTestGroupScan(); + + AccumuloGroupScan newScan = (AccumuloGroupScan) original.applyLimit(100); + + assertEquals(100, newScan.getMaxRecords()); + assertTrue("Limit should be marked as pushed down", newScan.isLimitPushedDown()); + } + + @Test + public void testApplyLimitDoesNotModifyOriginal() { + AccumuloGroupScan original = createTestGroupScan(); + int originalMaxRecords = original.getMaxRecords(); + + original.applyLimit(100); + + assertEquals("Original maxRecords should be unchanged", originalMaxRecords, original.getMaxRecords()); + assertFalse("Original should not have limit pushed down", original.isLimitPushedDown()); + } + + @Test + public void testApplyLimitWithMoreRestrictiveExisting() { + AccumuloStoragePlugin mockPlugin = Mockito.mock(AccumuloStoragePlugin.class); + AccumuloStoragePluginConfig mockConfig = Mockito.mock(AccumuloStoragePluginConfig.class); + Mockito.when(mockPlugin.getConfig()).thenReturn(mockConfig); + + AccumuloScanSpec scanSpec = new AccumuloScanSpec("test_table"); + // Create scan with limit already set to 50 + AccumuloGroupScan original = new AccumuloGroupScan("testUser", mockPlugin, scanSpec, null, 50); + + // Try to apply a higher limit + GroupScan newScan = original.applyLimit(100); + + // Should return null because existing limit is more restrictive + assertNull("Should return null when existing limit is more restrictive", newScan); + } + + @Test + public void testApplyLimitWithLessRestrictiveExisting() { + AccumuloStoragePlugin mockPlugin = Mockito.mock(AccumuloStoragePlugin.class); + AccumuloStoragePluginConfig mockConfig = Mockito.mock(AccumuloStoragePluginConfig.class); + Mockito.when(mockPlugin.getConfig()).thenReturn(mockConfig); + + AccumuloScanSpec scanSpec = new AccumuloScanSpec("test_table"); + // Create scan with limit already set to 100 + AccumuloGroupScan original = new AccumuloGroupScan("testUser", mockPlugin, scanSpec, null, 100); + + // Try to apply a lower limit + AccumuloGroupScan newScan = (AccumuloGroupScan) original.applyLimit(50); + + // Should return new scan with lower limit + assertNotNull("Should return new scan with more restrictive limit", newScan); + assertEquals(50, newScan.getMaxRecords()); + } + + @Test + public void testApplyLimitIsNotReappliedToItsOwnResult() { + // The planner rule keeps firing as long as applyLimit hands back a new scan, so + // re-applying the same limit must return null or planning never terminates. + AccumuloGroupScan original = createTestGroupScan(); + + AccumuloGroupScan limited = (AccumuloGroupScan) original.applyLimit(100); + + assertNull("Re-applying the same limit should return null", limited.applyLimit(100)); + } + + @Test + public void testApplyLimitZeroIsNotReapplied() { + // LIMIT 0 is the case that regressed: a zero limit must still be recognised as + // already pushed down. + AccumuloGroupScan original = createTestGroupScan(); + + AccumuloGroupScan limited = (AccumuloGroupScan) original.applyLimit(0); + + assertNotNull("A zero limit should still be pushed down", limited); + assertEquals(0, limited.getMaxRecords()); + assertNull("Re-applying a zero limit should return null", limited.applyLimit(0)); + } + + @Test + public void testApplyLimitPreservesOtherPushdowns() { + AccumuloGroupScan original = createTestGroupScan(); + original.setFilterPushedDown(true); + original.setProjectionPushedDown(true); + original.setSortPushedDown(true); + + AccumuloGroupScan newScan = (AccumuloGroupScan) original.applyLimit(100); + + assertTrue("Filter pushdown should be preserved", newScan.isFilterPushedDown()); + assertTrue("Projection pushdown should be preserved", newScan.isProjectionPushedDown()); + assertTrue("Sort pushdown should be preserved", newScan.isSortPushedDown()); + assertTrue("Limit should be marked as pushed down", newScan.isLimitPushedDown()); + } + + @Test + public void testApplyLimitPreservesScanSpec() { + AccumuloGroupScan original = createTestGroupScan(); + + AccumuloGroupScan newScan = (AccumuloGroupScan) original.applyLimit(100); + + assertNotNull("ScanSpec should be preserved", newScan.getScanSpec()); + assertEquals("test_table", newScan.getTableName()); + } + + @Test + public void testApplyLimitPreservesColumns() { + AccumuloGroupScan original = createTestGroupScan(); + ListThese tests verify that filter, projection, limit, and sort pushdowns + * work correctly with real Accumulo tables.
+ */ +public class AccumuloPushdownIntegrationTest extends BaseAccumuloTest { + + // ========================================================================= + // Filter Pushdown Tests + // ========================================================================= + + @Test + public void testFilterOnRowKeyEquals() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " WHERE row_key = 'row_001'"; + runAccumuloSQLVerifyCount(sql, 1); + } + + @Test + public void testFilterOnRowKeyGreaterThan() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " WHERE row_key > 'row_005'"; + runAccumuloSQLVerifyCount(sql, 5); // row_006 to row_010 + } + + @Test + public void testFilterOnRowKeyLessThan() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " WHERE row_key < 'row_004'"; + runAccumuloSQLVerifyCount(sql, 3); // row_001 to row_003 + } + + @Test + public void testFilterOnRowKeyRange() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " WHERE row_key >= 'row_003' AND row_key <= 'row_007'"; + runAccumuloSQLVerifyCount(sql, 5); // row_003 to row_007 + } + + @Test + public void testFilterOnRowKeyRangeLarge() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_LARGE) + " t" + + " WHERE row_key >= 'row_0100' AND row_key < 'row_0200'"; + runAccumuloSQLVerifyCount(sql, 100); // row_0100 to row_0199 + } + + @Test + public void testFilterOnRowKeyDisjunction() throws Exception { + // The pushed range spans both operands, so the filter must be kept in the plan + // to discard the rows in between. + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " WHERE row_key = 'row_001' OR row_key = 'row_009'"; + runAccumuloSQLVerifyCount(sql, 2); + } + + @Test + public void testFilterOnRowKeyDisjunctionOfRanges() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " WHERE row_key < 'row_003' OR row_key > 'row_008'"; + runAccumuloSQLVerifyCount(sql, 4); // row_001, row_002, row_009, row_010 + } + + @Test + public void testFilterOnColumnValue() throws Exception { + // Note: column value filters may not be pushed down, but should still work + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_USERS) + " t" + + " WHERE t.employment.company = 'Acme Corp'"; + runAccumuloSQLVerifyCount(sql, 7); // 7 users at Acme Corp + } + + // ========================================================================= + // Projection Pushdown Tests + // ========================================================================= + + @Test + public void testProjectionSingleColumn() throws Exception { + String sql = "SELECT t.cf.name FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t"; + runAccumuloSQLVerifyCount(sql, 10); + } + + @Test + public void testProjectionMultipleColumns() throws Exception { + String sql = "SELECT t.cf.name, t.cf.city FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t"; + runAccumuloSQLVerifyCount(sql, 10); + } + + @Test + public void testProjectionWithRowKey() throws Exception { + String sql = "SELECT row_key, t.personal.first_name FROM " + + fullTableName(AccumuloTestUtils.TEST_TABLE_USERS) + " t"; + runAccumuloSQLVerifyCount(sql, 20); + } + + @Test + public void testProjectionSingleColumnFamily() throws Exception { + String sql = "SELECT personal FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_USERS) + " t"; + runAccumuloSQLVerifyCount(sql, 20); + } + + // ========================================================================= + // Limit Pushdown Tests + // ========================================================================= + + @Test + public void testLimitSmall() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + " LIMIT 5"; + runAccumuloSQLVerifyCount(sql, 5); + } + + @Test + public void testLimitOnLargeTable() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_LARGE) + " t" + " LIMIT 50"; + runAccumuloSQLVerifyCount(sql, 50); + } + + @Test + public void testLimitOne() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + " LIMIT 1"; + runAccumuloSQLVerifyCount(sql, 1); + } + + @Test + public void testLimitLargerThanTable() throws Exception { + // Limit larger than table size should return all rows + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + " LIMIT 100"; + runAccumuloSQLVerifyCount(sql, 10); + } + + // ========================================================================= + // Sort Pushdown Tests + // ========================================================================= + + @Test + public void testOrderByRowKeyAsc() throws Exception { + String sql = "SELECT row_key FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " ORDER BY row_key ASC"; + runAccumuloSQLVerifyCount(sql, 10); + } + + @Test + public void testOrderByRowKeyDesc() throws Exception { + String sql = "SELECT row_key FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " ORDER BY row_key DESC"; + runAccumuloSQLVerifyCount(sql, 10); + } + + @Test + public void testOrderByRowKeyWithLimit() throws Exception { + String sql = "SELECT row_key FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_LARGE) + " t" + + " ORDER BY row_key ASC LIMIT 10"; + runAccumuloSQLVerifyCount(sql, 10); + } + + // ========================================================================= + // Combined Pushdown Tests + // ========================================================================= + + @Test + public void testFilterAndProjection() throws Exception { + String sql = "SELECT row_key, t.cf.name FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " WHERE row_key > 'row_005'"; + runAccumuloSQLVerifyCount(sql, 5); + } + + @Test + public void testFilterAndLimit() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " WHERE row_key > 'row_002' LIMIT 3"; + runAccumuloSQLVerifyCount(sql, 3); + } + + @Test + public void testProjectionAndLimit() throws Exception { + String sql = "SELECT row_key, t.cf.name FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " LIMIT 5"; + runAccumuloSQLVerifyCount(sql, 5); + } + + @Test + public void testFilterProjectionAndLimit() throws Exception { + String sql = "SELECT row_key, t.cf.name, t.cf.city FROM " + + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " WHERE row_key >= 'row_003' LIMIT 4"; + runAccumuloSQLVerifyCount(sql, 4); + } + + @Test + public void testFilterAndSort() throws Exception { + String sql = "SELECT row_key FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " WHERE row_key < 'row_006' ORDER BY row_key ASC"; + runAccumuloSQLVerifyCount(sql, 5); + } + + @Test + public void testSortAndLimit() throws Exception { + String sql = "SELECT row_key FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " ORDER BY row_key ASC LIMIT 3"; + runAccumuloSQLVerifyCount(sql, 3); + } + + @Test + public void testAllPushdownsCombined() throws Exception { + String sql = "SELECT row_key, t.cf.name FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " WHERE row_key >= 'row_002' AND row_key <= 'row_009'" + + " ORDER BY row_key ASC LIMIT 5"; + runAccumuloSQLVerifyCount(sql, 5); + } + + // ========================================================================= + // Edge Case Tests + // ========================================================================= + + @Test + public void testFilterNoResults() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " WHERE row_key = 'nonexistent'"; + runAccumuloSQLVerifyCount(sql, 0); + } + + @Test + public void testFilterOutOfRange() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + + " WHERE row_key > 'zzz'"; + runAccumuloSQLVerifyCount(sql, 0); + } + + @Test + public void testLimitZero() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1) + " t" + " LIMIT 0"; + runAccumuloSQLVerifyCount(sql, 0); + } +} diff --git a/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloResultVerificationTest.java b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloResultVerificationTest.java new file mode 100644 index 00000000000..b1dd39e7755 --- /dev/null +++ b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloResultVerificationTest.java @@ -0,0 +1,347 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Test; + +/** + * End-to-end tests that verify the actual values Drill returns from Accumulo, + * not just the number of rows. + * + *Accumulo row keys and values are surfaced to Drill as VARBINARY, so the queries + * here decode them with {@code CONVERT_FROM(..., 'UTF8')} before comparing against + * the data written by {@link AccumuloTestUtils}.
+ */ +public class AccumuloResultVerificationTest extends BaseAccumuloTest { + + // ========================================================================= + // Full table content + // ========================================================================= + + @Test + public void testAllRowsAndValuesFromTable1() throws Exception { + String sql = "SELECT " + utf8("row_key", "row_key") + ", " + + utf8("t.cf.name", "name") + ", " + + utf8("t.cf.age", "age") + ", " + + utf8("t.cf.city", "city") + + fromTable(AccumuloTestUtils.TEST_TABLE_1) + + " ORDER BY row_key"; + + testBuilder() + .sqlQuery(sql) + .ordered() + .baselineColumns("row_key", "name", "age", "city") + .baselineValues("row_001", "Alice", "30", "New York") + .baselineValues("row_002", "Bob", "25", "Los Angeles") + .baselineValues("row_003", "Charlie", "35", "Chicago") + .baselineValues("row_004", "Diana", "28", "Houston") + .baselineValues("row_005", "Eve", "32", "Phoenix") + .baselineValues("row_006", "Frank", "45", "Philadelphia") + .baselineValues("row_007", "Grace", "29", "San Antonio") + .baselineValues("row_008", "Henry", "38", "San Diego") + .baselineValues("row_009", "Ivy", "26", "Dallas") + .baselineValues("row_010", "Jack", "41", "San Jose") + .go(); + } + + @Test + public void testValuesAcrossMultipleColumnFamilies() throws Exception { + String sql = "SELECT " + utf8("row_key", "row_key") + ", " + + utf8("t.personal.first_name", "first_name") + ", " + + utf8("t.personal.last_name", "last_name") + ", " + + utf8("t.contact.email", "email") + ", " + + utf8("t.employment.company", "company") + ", " + + utf8("t.employment.salary", "salary") + + fromTable(AccumuloTestUtils.TEST_TABLE_USERS) + + " WHERE row_key IN ('user_001', 'user_014', 'user_020')" + + " ORDER BY row_key"; + + testBuilder() + .sqlQuery(sql) + .ordered() + .baselineColumns("row_key", "first_name", "last_name", "email", "company", "salary") + .baselineValues("user_001", "John", "Doe", "john.doe@email.com", "Acme Corp", "75000") + .baselineValues("user_014", "Laura", "White", "laura.w@email.com", "TechCo", "125000") + .baselineValues("user_020", "Rachel", "Clark", "rachel.c@email.com", "TechCo", "99000") + .go(); + } + + // ========================================================================= + // Filter pushdown: verify the correct rows come back, not just the count + // ========================================================================= + + @Test + public void testRowKeyEqualsReturnsMatchingRow() throws Exception { + String sql = "SELECT " + utf8("row_key", "row_key") + ", " + utf8("t.cf.name", "name") + + fromTable(AccumuloTestUtils.TEST_TABLE_1) + + " WHERE row_key = 'row_003'"; + + testBuilder() + .sqlQuery(sql) + .unOrdered() + .baselineColumns("row_key", "name") + .baselineValues("row_003", "Charlie") + .go(); + } + + @Test + public void testRowKeyRangeReturnsExactRows() throws Exception { + String sql = "SELECT " + utf8("row_key", "row_key") + ", " + utf8("t.cf.city", "city") + + fromTable(AccumuloTestUtils.TEST_TABLE_1) + + " WHERE row_key >= 'row_003' AND row_key <= 'row_005'" + + " ORDER BY row_key"; + + testBuilder() + .sqlQuery(sql) + .ordered() + .baselineColumns("row_key", "city") + .baselineValues("row_003", "Chicago") + .baselineValues("row_004", "Houston") + .baselineValues("row_005", "Phoenix") + .go(); + } + + @Test + public void testRowKeyGreaterThanReturnsExactRows() throws Exception { + String sql = "SELECT " + utf8("row_key", "row_key") + + fromTable(AccumuloTestUtils.TEST_TABLE_1) + + " WHERE row_key > 'row_007'" + + " ORDER BY row_key"; + + testBuilder() + .sqlQuery(sql) + .ordered() + .baselineColumns("row_key") + .baselineValues("row_008") + .baselineValues("row_009") + .baselineValues("row_010") + .go(); + } + + @Test + public void testValueFilterReturnsMatchingRows() throws Exception { + String sql = "SELECT " + utf8("row_key", "row_key") + + fromTable(AccumuloTestUtils.TEST_TABLE_USERS) + + " WHERE CONVERT_FROM(t.employment.title, 'UTF8') = 'Director'" + + " ORDER BY row_key"; + + testBuilder() + .sqlQuery(sql) + .ordered() + .baselineColumns("row_key") + .baselineValues("user_004") + .baselineValues("user_014") + .go(); + } + + @Test + public void testRowKeyRangeOnLargeTableBoundaries() throws Exception { + // Exercises a range that spans many rows: verify both endpoints and the count. + String sql = "SELECT MIN(rk) AS min_rk, MAX(rk) AS max_rk, COUNT(*) AS cnt FROM (" + + " SELECT " + utf8("row_key", "rk") + + " " + fromTable(AccumuloTestUtils.TEST_TABLE_LARGE) + + " WHERE row_key >= 'row_0100' AND row_key < 'row_0200')"; + + testBuilder() + .sqlQuery(sql) + .unOrdered() + .baselineColumns("min_rk", "max_rk", "cnt") + .baselineValues("row_0100", "row_0199", 100L) + .go(); + } + + // ========================================================================= + // Sort and limit + // ========================================================================= + + @Test + public void testOrderByRowKeyDescReturnsRowsInOrder() throws Exception { + String sql = "SELECT " + utf8("row_key", "row_key") + + fromTable(AccumuloTestUtils.TEST_TABLE_1) + + " ORDER BY row_key DESC LIMIT 3"; + + testBuilder() + .sqlQuery(sql) + .ordered() + .baselineColumns("row_key") + .baselineValues("row_010") + .baselineValues("row_009") + .baselineValues("row_008") + .go(); + } + + @Test + public void testOrderByWithLimitOnLargeTable() throws Exception { + String sql = "SELECT " + utf8("row_key", "row_key") + ", " + utf8("t.data.value", "value") + + fromTable(AccumuloTestUtils.TEST_TABLE_LARGE) + + " ORDER BY row_key LIMIT 3"; + + testBuilder() + .sqlQuery(sql) + .ordered() + .baselineColumns("row_key", "value") + .baselineValues("row_0001", "1") + .baselineValues("row_0002", "2") + .baselineValues("row_0003", "3") + .go(); + } + + @Test + public void testLimitReturnsDistinctRowsFromTheTable() throws Exception { + // A pushed-down limit must return whole, distinct rows rather than repeating or + // truncating them, so check the returned keys against the full key set. + String sql = "SELECT " + utf8("row_key", "row_key") + + fromTable(AccumuloTestUtils.TEST_TABLE_LARGE) + " LIMIT 25"; + + ListDrill serializes physical operators to JSON when it distributes fragments to + * other Drillbits, so anything that survives planning must survive a JSON round trip. + * These tests cover both the whole-plan path (plan the query, serialize it, then submit + * the serialized plan and check the results) and a direct round trip of + * {@link AccumuloSubScan} through Drill's {@link PhysicalPlanReader}.
+ */ +public class AccumuloSerDeTest extends BaseAccumuloTest { + + // ========================================================================= + // Whole-plan round trip: plan -> JSON -> execute -> verify results + // ========================================================================= + + @Test + public void testSerDeSelectStar() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_1); + String plan = queryBuilder().sql(sql).explainJson(); + + assertTrue("Plan should contain the Accumulo scan", plan.contains("accumulo-scan")); + assertEquals(10, queryBuilder().physical(plan).run().recordCount()); + } + + @Test + public void testSerDePreservesValues() throws Exception { + String sql = "SELECT " + utf8("row_key", "row_key") + ", " + utf8("t.cf.name", "name") + + fromTable(AccumuloTestUtils.TEST_TABLE_1) + + " WHERE row_key >= 'row_008' ORDER BY row_key"; + String plan = queryBuilder().sql(sql).explainJson(); + + assertEquals( + Arrays.asList( + Arrays.asList("row_008", "Henry"), + Arrays.asList("row_009", "Ivy"), + Arrays.asList("row_010", "Jack")), + readStringsFromPlan(plan)); + } + + @Test + public void testSerDePreservesRowRangePushdown() throws Exception { + String sql = "SELECT row_key FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_LARGE) + + " WHERE row_key >= 'row_0100' AND row_key < 'row_0200'"; + String plan = queryBuilder().sql(sql).explainJson(); + + // The deserialized plan must scan the same range, not the whole table. + assertEquals(100, queryBuilder().physical(plan).run().recordCount()); + } + + @Test + public void testSerDePreservesLimitPushdown() throws Exception { + String sql = "SELECT * FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_LARGE) + " LIMIT 17"; + String plan = queryBuilder().sql(sql).explainJson(); + + assertEquals(17, queryBuilder().physical(plan).run().recordCount()); + } + + @Test + public void testSerDeAggregate() throws Exception { + String sql = "SELECT COUNT(*) FROM " + fullTableName(AccumuloTestUtils.TEST_TABLE_USERS); + String plan = queryBuilder().sql(sql).explainJson(); + + assertEquals(20L, queryBuilder().physical(plan).singletonLong()); + } + + @Test + public void testFragmentSerDe() throws Exception { + // A slice target of 1 forces the plan to be split into fragments, which are + // serialized individually before being handed to the executor. + client.alterSession(ExecConstants.SLICE_TARGET, 1); + try { + String sql = "SELECT CONVERT_FROM(t.employment.company, 'UTF8') AS company, COUNT(*) AS cnt" + + fromTable(AccumuloTestUtils.TEST_TABLE_USERS) + + " GROUP BY CONVERT_FROM(t.employment.company, 'UTF8')"; + String plan = queryBuilder().sql(sql).explainJson(); + + ListTable structure:
+ *Table structure:
+ *Table structure:
+ *This exercises the record reader's handling of columns that are absent from + * some rows: every missing qualifier must come back as NULL rather than shifting + * values between rows.
+ * + *Table structure:
+ *This suite includes all unit tests for the Accumulo plugin components. + * Integration tests requiring MiniAccumuloCluster will be added in later phases.
+ */ +@RunWith(Suite.class) +@Suite.SuiteClasses({ + // Phase 1: Core components + AccumuloStoragePluginConfigTest.class, + AccumuloScanSpecTest.class, + // Phase 2: Schema discovery + AccumuloColumnTypeTest.class, + TableSchemaTest.class, + // Phase 3: Basic scan capability + DrillAccumuloConstantsTest.class, + AccumuloTypeConverterTest.class, + // Phase 4: Filter pushdown + AccumuloFilterBuilderTest.class, + // Phase 5: Projection pushdown + AccumuloProjectionPushdownTest.class, + // Phase 6: Limit pushdown + AccumuloLimitPushdownTest.class, + // Phase 7: Sort pushdown + AccumuloSortPushdownTest.class, + // Phase 8: Kerberos authentication + AccumuloKerberosConfigTest.class, + DelegationTokenInfoTest.class +}) +public class AccumuloTestsSuite { + // Test suite - no implementation needed +} diff --git a/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloTypeConverterTest.java b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloTypeConverterTest.java new file mode 100644 index 00000000000..0f1f3d34d9a --- /dev/null +++ b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/AccumuloTypeConverterTest.java @@ -0,0 +1,270 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; + +import org.apache.drill.exec.store.accumulo.schema.AccumuloColumnType; +import org.apache.drill.test.BaseTest; +import org.junit.Test; + +/** + * Unit tests for AccumuloTypeConverter. + */ +public class AccumuloTypeConverterTest extends BaseTest { + + @Test + public void testConvertVarchar() { + byte[] bytes = "hello world".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.VARCHAR); + assertEquals("hello world", result); + } + + @Test + public void testConvertVarcharEmpty() { + byte[] bytes = new byte[0]; + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.VARCHAR); + assertNull(result); + } + + @Test + public void testConvertVarcharNull() { + Object result = AccumuloTypeConverter.convert(null, AccumuloColumnType.VARCHAR); + assertNull(result); + } + + @Test + public void testConvertIntegerFromString() { + byte[] bytes = "42".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.INT); + assertEquals(42, result); + } + + @Test + public void testConvertIntegerNegative() { + byte[] bytes = "-123".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.INTEGER); + assertEquals(-123, result); + } + + @Test + public void testConvertIntegerFromBinary() { + // Use a value that produces non-ASCII bytes so it falls through to binary parsing + ByteBuffer buffer = ByteBuffer.allocate(4); + buffer.putInt(0x80000001); // Has high bit set, produces non-printable chars + Object result = AccumuloTypeConverter.convert(buffer.array(), AccumuloColumnType.INT); + assertEquals(0x80000001, result); + } + + @Test + public void testConvertIntegerInvalid() { + byte[] bytes = "not a number".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.INT); + assertNull(result); + } + + @Test + public void testConvertLongFromString() { + byte[] bytes = "9223372036854775807".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.BIGINT); + assertEquals(Long.MAX_VALUE, result); + } + + @Test + public void testConvertLongFromBinary() { + ByteBuffer buffer = ByteBuffer.allocate(8); + buffer.putLong(123456789L); + Object result = AccumuloTypeConverter.convert(buffer.array(), AccumuloColumnType.LONG); + assertEquals(123456789L, result); + } + + @Test + public void testConvertFloat() { + byte[] bytes = "3.14".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.FLOAT); + assertEquals(3.14f, (Float) result, 0.001); + } + + @Test + public void testConvertFloatFromBinary() { + ByteBuffer buffer = ByteBuffer.allocate(4); + buffer.putFloat(3.14f); + Object result = AccumuloTypeConverter.convert(buffer.array(), AccumuloColumnType.FLOAT); + assertEquals(3.14f, (Float) result, 0.001); + } + + @Test + public void testConvertDouble() { + byte[] bytes = "3.14159265359".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.DOUBLE); + assertEquals(3.14159265359, (Double) result, 0.00000000001); + } + + @Test + public void testConvertDoubleFromBinary() { + ByteBuffer buffer = ByteBuffer.allocate(8); + buffer.putDouble(3.14159265359); + Object result = AccumuloTypeConverter.convert(buffer.array(), AccumuloColumnType.DOUBLE); + assertEquals(3.14159265359, (Double) result, 0.00000000001); + } + + @Test + public void testConvertDecimal() { + byte[] bytes = "123456.789".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.DECIMAL); + assertEquals(new BigDecimal("123456.789"), result); + } + + @Test + public void testConvertBooleanTrue() { + byte[] bytes = "true".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.BOOLEAN); + assertTrue((Boolean) result); + } + + @Test + public void testConvertBooleanFalse() { + byte[] bytes = "false".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.BOOLEAN); + assertFalse((Boolean) result); + } + + @Test + public void testConvertBooleanOne() { + byte[] bytes = "1".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.BOOLEAN); + assertTrue((Boolean) result); + } + + @Test + public void testConvertBooleanZero() { + byte[] bytes = "0".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.BOOLEAN); + assertFalse((Boolean) result); + } + + @Test + public void testConvertBooleanYes() { + byte[] bytes = "YES".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.BOOLEAN); + assertTrue((Boolean) result); + } + + @Test + public void testConvertBooleanBinary() { + byte[] bytes = new byte[]{1}; + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.BOOLEAN); + assertTrue((Boolean) result); + + bytes = new byte[]{0}; + result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.BOOLEAN); + assertFalse((Boolean) result); + } + + @Test + public void testConvertDateIso() { + byte[] bytes = "2024-01-15".getBytes(StandardCharsets.UTF_8); + Long result = (Long) AccumuloTypeConverter.convert(bytes, AccumuloColumnType.DATE); + // 2024-01-15 00:00:00 UTC + assertEquals(1705276800000L, result.longValue()); + } + + @Test + public void testConvertTimeIso() { + byte[] bytes = "12:30:45".getBytes(StandardCharsets.UTF_8); + Integer result = (Integer) AccumuloTypeConverter.convert(bytes, AccumuloColumnType.TIME); + // 12:30:45 = 12*3600*1000 + 30*60*1000 + 45*1000 = 45045000 ms + assertEquals(45045000, result.intValue()); + } + + @Test + public void testConvertTimestampIso() { + byte[] bytes = "2024-01-15T12:30:45Z".getBytes(StandardCharsets.UTF_8); + Long result = (Long) AccumuloTypeConverter.convert(bytes, AccumuloColumnType.TIMESTAMP); + assertEquals(1705321845000L, result.longValue()); + } + + @Test + public void testConvertTimestampEpoch() { + byte[] bytes = "1705321845000".getBytes(StandardCharsets.UTF_8); + Long result = (Long) AccumuloTypeConverter.convert(bytes, AccumuloColumnType.TIMESTAMP); + assertEquals(1705321845000L, result.longValue()); + } + + @Test + public void testConvertVarbinary() { + byte[] bytes = new byte[]{0x01, 0x02, 0x03, (byte) 0xFF}; + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.VARBINARY); + assertArrayEquals(bytes, (byte[]) result); + } + + @Test + public void testConvertAny() { + byte[] bytes = "some value".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.ANY); + assertEquals("some value", result); + } + + @Test + public void testToDisplayStringPrintable() { + byte[] bytes = "Hello World".getBytes(StandardCharsets.UTF_8); + String result = AccumuloTypeConverter.toDisplayString(bytes); + assertEquals("Hello World", result); + } + + @Test + public void testToDisplayStringBinary() { + byte[] bytes = new byte[]{0x01, 0x02, 0x03}; + String result = AccumuloTypeConverter.toDisplayString(bytes); + assertEquals("0x010203", result); + } + + @Test + public void testToDisplayStringNull() { + String result = AccumuloTypeConverter.toDisplayString(null); + assertEquals("null", result); + } + + @Test + public void testToDisplayStringEmpty() { + String result = AccumuloTypeConverter.toDisplayString(new byte[0]); + assertEquals("", result); + } + + @Test + public void testConvertShort() { + byte[] bytes = "32767".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.SMALLINT); + assertEquals((short) 32767, result); + } + + @Test + public void testConvertByte() { + byte[] bytes = "127".getBytes(StandardCharsets.UTF_8); + Object result = AccumuloTypeConverter.convert(bytes, AccumuloColumnType.TINYINT); + assertEquals((byte) 127, result); + } +} diff --git a/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/BaseAccumuloTest.java b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/BaseAccumuloTest.java new file mode 100644 index 00000000000..08c7f0644ab --- /dev/null +++ b/contrib/storage-accumulo/src/test/java/org/apache/drill/exec/store/accumulo/BaseAccumuloTest.java @@ -0,0 +1,153 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.drill.exec.physical.rowSet.DirectRowSet; +import org.apache.drill.exec.physical.rowSet.RowSetReader; +import org.apache.drill.exec.store.StoragePluginRegistry; +import org.apache.drill.exec.vector.accessor.ScalarReader; +import org.apache.drill.test.ClusterFixture; +import org.apache.drill.test.ClusterTest; +import org.apache.drill.test.QueryRowSetIterator; +import org.junit.AfterClass; +import org.junit.BeforeClass; + +/** + * Base class for Accumulo integration tests. + * + *This class sets up the Drill test cluster and registers the Accumulo storage plugin + * configured to connect to the MiniAccumuloCluster.
+ */ +public class BaseAccumuloTest extends ClusterTest { + + public static final String ACCUMULO_STORAGE_PLUGIN_NAME = "accumulo"; + + protected static AccumuloStoragePlugin storagePlugin; + protected static AccumuloStoragePluginConfig storagePluginConfig; + + @BeforeClass + public static void setupAccumuloTestCluster() throws Exception { + // Initialize the MiniAccumuloCluster + boolean isManaged = Boolean.parseBoolean(System.getProperty("drill.accumulo.tests.managed", "true")); + AccumuloIntegrationTestsSuite.configure(isManaged, true); + AccumuloIntegrationTestsSuite.initCluster(); + + // Start the Drill test cluster + startCluster(ClusterFixture.builder(dirTestWatcher)); + + // Register Accumulo storage plugin + StoragePluginRegistry pluginRegistry = cluster.drillbit().getContext().getStorage(); + storagePluginConfig = new AccumuloStoragePluginConfig( + AccumuloIntegrationTestsSuite.getZooKeepers(), + AccumuloIntegrationTestsSuite.getInstanceName(), + AccumuloIntegrationTestsSuite.getRootUser(), + AccumuloIntegrationTestsSuite.getRootPassword() + ); + storagePluginConfig.setEnabled(true); + + pluginRegistry.put(ACCUMULO_STORAGE_PLUGIN_NAME, storagePluginConfig); + storagePlugin = (AccumuloStoragePlugin) pluginRegistry.getPlugin(ACCUMULO_STORAGE_PLUGIN_NAME); + } + + @AfterClass + public static void tearDownAccumuloTestCluster() throws Exception { + AccumuloIntegrationTestsSuite.tearDownCluster(); + } + + /** + * Runs a SQL query and verifies the row count. Pass {@code -1} to skip the check. + */ + protected void runAccumuloSQLVerifyCount(String sql, int expectedRowCount) throws Exception { + long rowCount = queryBuilder().sql(sql).run().recordCount(); + if (expectedRowCount != -1) { + assertEquals(expectedRowCount, rowCount); + } + } + + /** + * Returns the fully qualified table name for Drill queries. + * + * @param tableName the Accumulo table name + * @return the fully qualified name like "accumulo.`tableName`" + */ + protected String fullTableName(String tableName) { + return ACCUMULO_STORAGE_PLUGIN_NAME + ".`" + tableName + "`"; + } + + /** + * Returns a {@code FROM} clause that aliases the table as {@code t}. Referring to a + * qualifier inside a column family requires the table alias ({@code t.cf.name}), + * the same as for the HBase plugin. + */ + protected String fromTable(String tableName) { + return " FROM " + fullTableName(tableName) + " t"; + } + + /** + * Wraps a column reference in a {@code CONVERT_FROM(..., 'UTF8')} call. Accumulo row + * keys and values are surfaced to Drill as VARBINARY, so they must be decoded before + * they can be compared against string baselines. + * + * @param column the column reference, e.g. {@code row_key} or {@code cf.name} + * @param alias the alias to give the decoded column + */ + protected static String utf8(String column, String alias) { + return "CONVERT_FROM(" + column + ", 'UTF8') AS " + alias; + } + + /** + * Runs a query and returns its results as rows of strings, with {@code null} for + * NULL values. Reading the values back as strings keeps the assertions independent + * of whether a column comes back as required or nullable. + */ + protected ListEvery batch is read, so this stays correct for queries that return their results + * across several batches, and each row set is released as it is consumed.
+ */ + protected static List