[CALCITE-5681] Support authorization via GRANT and REVOKE DDL commands

core/src/main/codegen/default_config.fmpp

@@ -429,6 +429,12 @@ parser: {
   truncateStatementParserMethods: [
   ]
 
+  # List of methods for parsing extensions to "GRANT" or "REVOKE" calls.
+  # Each must accept arguments "(SqlParserPos pos)".
+  # Example: "SqlGrant".
+  privilegeStatementParserMethods: [
+  ]
+
   # Binary operators tokens.
   # Example: "< INFIX_CAST: \"::\" >".
   binaryOperatorsTokens: [

core/src/main/codegen/templates/Parser.jj

@@ -1161,6 +1161,10 @@ SqlNode SqlStmt() :
         stmt = SqlTruncate()
     |
 </#if>
+<#list (parser.privilegeStatementParserMethods!default.parser.privilegeStatementParserMethods) as method>
+        stmt = ${method}
+    |
+</#list>
         stmt = OrderedQueryOrExpr(ExprContext.ACCEPT_QUERY)
     |
         stmt = SqlExplain()

core/src/main/java/org/apache/calcite/access/CalcitePrincipal.java

@@ -0,0 +1,69 @@
+/*
+ * 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.calcite.access;
+
+import org.apache.commons.lang.StringUtils;
+
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.security.Principal;
+import java.util.Locale;
+
+import static org.apache.calcite.runtime.Utilities.hash;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * CalcitePrincipal is a simple implementation of {@link Principal}
+ * that represents a user. CalcitePrincipal ignore the case of the
+ * name when comparing two principals.
+ */
+public class CalcitePrincipal implements Principal, Comparable<CalcitePrincipal> {
+
+  private final String name;
+
+  public CalcitePrincipal(String name) {
+    this.name = requireNonNull(name, "name").toUpperCase(Locale.ROOT);
+  }
+
+  @Override public String getName() {
+    return this.name;
+  }
+
+  @Override public String toString() {
+    return this.name;
+  }
+
+  @Override public boolean equals(@Nullable Object o) {
+    if (this == o) {
+      return true;
+    } else if (o != null && this.getClass() == o.getClass()) {
+      CalcitePrincipal that = (CalcitePrincipal) o;
+      return StringUtils.equalsIgnoreCase(this.name, that.name);
+    } else {
+      return false;
+    }
+  }
+
+  @Override public int hashCode() {
+    return hash(this.name);
+  }
+
+  @Override public int compareTo(CalcitePrincipal o) {
+    return String.CASE_INSENSITIVE_ORDER.compare(this.name, o.name);
+  }
+}

core/src/main/java/org/apache/calcite/access/CalcitePrincipalFactory.java

@@ -0,0 +1,48 @@
+/*
+ * 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.calcite.access;
+
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Factory for {@link CalcitePrincipal}.
+ */
+public class CalcitePrincipalFactory {
+
+  public static final CalcitePrincipalFactory INSTANCE = new CalcitePrincipalFactory();
+
+  private final LoadingCache<String, CalcitePrincipal> cache =
+      CacheBuilder.newBuilder().build(CacheLoader.from(CalcitePrincipalFactory::createInternal));
+
+  private CalcitePrincipalFactory() {
+  }
+
+  private static CalcitePrincipal createInternal(String name) {
+    return new CalcitePrincipal(name);
+  }
+
+  public static @Nullable CalcitePrincipal create(@Nullable String name) {
+    if (name == null) {
+      return null;
+    }
+    return INSTANCE.cache.getUnchecked(name);
+  }
+}

core/src/main/java/org/apache/calcite/access/package-info.java

@@ -0,0 +1,28 @@
+/*
+ * 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.
+ */
+
+/**
+ * Calcite Access Control.
+ */
+@DefaultQualifier(value = NonNull.class, locations = TypeUseLocation.FIELD)
+@DefaultQualifier(value = NonNull.class, locations = TypeUseLocation.PARAMETER)
+@DefaultQualifier(value = NonNull.class, locations = TypeUseLocation.RETURN)
+package org.apache.calcite.access;
+
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.framework.qual.DefaultQualifier;
+import org.checkerframework.framework.qual.TypeUseLocation;

core/src/main/java/org/apache/calcite/config/CalciteConnectionConfig.java

@@ -122,4 +122,10 @@ public interface CalciteConnectionConfig extends ConnectionConfig {
    * {@code defaultMetaColumnFactory} is not null, the result is never null. */
   <T> @PolyNull T metaColumnFactory(Class<T> metaColumnFactoryClass,
       @PolyNull T defaultMetaColumnFactory);
+
+  /** Returns the value of {@link CalciteConnectionProperty#ACCESS_CONTROL}. */
+  boolean accessControl();
+
+  /** Returns the value of {@link CalciteConnectionProperty#USER}. */
+  @Nullable String user();
 }

core/src/main/java/org/apache/calcite/config/CalciteConnectionConfigImpl.java

@@ -228,4 +228,13 @@ public boolean isSet(CalciteConnectionProperty property) {
     return CalciteConnectionProperty.META_COLUMN_FACTORY.wrap(properties)
         .getPlugin(metaColumnFactoryClass, defaultMetaColumnFactory);
   }
+
+  @Override public boolean accessControl() {
+    return CalciteConnectionProperty.ACCESS_CONTROL.wrap(properties)
+        .getBoolean();
+  }
+
+  @Override public @Nullable String user() {
+    return CalciteConnectionProperty.USER.wrap(properties).getString();
+  }
 }

core/src/main/java/org/apache/calcite/config/CalciteConnectionProperty.java

@@ -184,7 +184,13 @@ public enum CalciteConnectionProperty implements ConnectionProperty {
   LENIENT_OPERATOR_LOOKUP("lenientOperatorLookup", Type.BOOLEAN, false, false),
 
   /** Whether to enable top-down optimization in Volcano planner. */
-  TOPDOWN_OPT("topDownOpt", Type.BOOLEAN, CalciteSystemProperty.TOPDOWN_OPT.value(), false);
+  TOPDOWN_OPT("topDownOpt", Type.BOOLEAN, CalciteSystemProperty.TOPDOWN_OPT.value(), false),
+
+  /** Whether to enable access control, default false. */
+  ACCESS_CONTROL("accessControl", Type.BOOLEAN, false, false),
+
+  /** User name. */
+  USER("user", Type.STRING, null, false);
 
   private final String camelName;
   private final Type type;

core/src/main/java/org/apache/calcite/jdbc/CachingCalciteSchema.java

@@ -22,6 +22,7 @@
 import org.apache.calcite.schema.SchemaVersion;
 import org.apache.calcite.schema.Table;
 import org.apache.calcite.schema.TableMacro;
+import org.apache.calcite.sql.SqlAccessType;
 import org.apache.calcite.util.NameMap;
 import org.apache.calcite.util.NameMultimap;
 import org.apache.calcite.util.NameSet;
@@ -35,8 +36,10 @@
 
 import org.checkerframework.checker.nullness.qual.Nullable;
 
+import java.security.Principal;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 import static org.apache.calcite.linq4j.Nullness.castNonNull;
@@ -56,7 +59,7 @@ class CachingCalciteSchema extends CalciteSchema {
   /** Creates a CachingCalciteSchema. */
   CachingCalciteSchema(@Nullable CalciteSchema parent, Schema schema,
       String name) {
-    this(parent, schema, name, null, null, null, null, null, null, null, null);
+    this(parent, schema, name, null, null, null, null, null, null, null, null, null);
   }
 
   @SuppressWarnings({"argument.type.incompatible", "return.type.incompatible"})
@@ -69,9 +72,10 @@ private CachingCalciteSchema(@Nullable CalciteSchema parent, Schema schema,
       @Nullable NameMultimap<FunctionEntry> functionMap,
       @Nullable NameSet functionNames,
       @Nullable NameMap<FunctionEntry> nullaryFunctionMap,
+      @Nullable NameMap<Map<Principal, SqlAccessType>> privilegeMap,
       @Nullable List<? extends List<String>> path) {
     super(parent, schema, name, subSchemaMap, tableMap, latticeMap, typeMap,
-        functionMap, functionNames, nullaryFunctionMap, path);
+        functionMap, functionNames, nullaryFunctionMap, privilegeMap, path);
     this.implicitSubSchemaCache =
         new AbstractCached<SubSchemaCache>() {
           @Override public SubSchemaCache build() {
@@ -260,7 +264,7 @@ private CachingCalciteSchema(@Nullable CalciteSchema parent, Schema schema,
     CalciteSchema snapshot =
         new CachingCalciteSchema(parent, schema.snapshot(version), name, null,
             tableMap, latticeMap, typeMap,
-            functionMap, functionNames, nullaryFunctionMap, getPath());
+            functionMap, functionNames, nullaryFunctionMap, privilegeMap, getPath());
     for (CalciteSchema subSchema : subSchemaMap.map().values()) {
       CalciteSchema subSchemaSnapshot = subSchema.snapshot(snapshot, version);
       snapshot.subSchemaMap.put(subSchema.name, subSchemaSnapshot);

core/src/main/java/org/apache/calcite/jdbc/CalciteConnectionImpl.java

@@ -427,7 +427,8 @@ static class DataContextImpl implements DataContext {
               .deriveTimeFrameSet(TimeFrames.CORE);
       final long localOffset = timeZone.getOffset(time);
       final long currentOffset = localOffset;
-      final String user = "sa";
+      final String user = connection.config().user() == null
+          ? "sa" : requireNonNull(connection.config().user());
       final String systemUser = System.getProperty("user.name");
       final String localeName = connection.config().locale();
       final Locale locale = localeName != null

core/src/main/java/org/apache/calcite/jdbc/CalciteSchema.java

@@ -29,6 +29,7 @@
 import org.apache.calcite.schema.Wrapper;
 import org.apache.calcite.schema.impl.MaterializedViewTable;
 import org.apache.calcite.schema.impl.StarTable;
+import org.apache.calcite.sql.SqlAccessType;
 import org.apache.calcite.util.NameMap;
 import org.apache.calcite.util.NameMultimap;
 import org.apache.calcite.util.NameSet;
@@ -40,8 +41,10 @@
 
 import org.checkerframework.checker.nullness.qual.Nullable;
 
+import java.security.Principal;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.NavigableMap;
@@ -71,6 +74,9 @@ public abstract class CalciteSchema {
   protected final NameSet functionNames;
   protected final NameMap<FunctionEntry> nullaryFunctionMap;
   protected final NameMap<CalciteSchema> subSchemaMap;
+
+  // tableName -> (user, access)
+  protected final NameMap<Map<Principal, SqlAccessType>> privilegeMap;
   private @Nullable List<? extends List<String>> path;
 
   protected CalciteSchema(@Nullable CalciteSchema parent, Schema schema,
@@ -82,6 +88,7 @@ protected CalciteSchema(@Nullable CalciteSchema parent, Schema schema,
       @Nullable NameMultimap<FunctionEntry> functionMap,
       @Nullable NameSet functionNames,
       @Nullable NameMap<FunctionEntry> nullaryFunctionMap,
+      @Nullable NameMap<Map<Principal, SqlAccessType>> privilegeMap,
       @Nullable List<? extends List<String>> path) {
     this.parent = parent;
     this.schema = schema;
@@ -106,6 +113,7 @@ protected CalciteSchema(@Nullable CalciteSchema parent, Schema schema,
     } else {
       this.typeMap = typeMap;
     }
+    this.privilegeMap = privilegeMap != null ? privilegeMap : new NameMap<>();
     this.path = path;
   }
 
@@ -543,6 +551,47 @@ public boolean removeType(String name) {
     return typeMap.remove(name) != null;
   }
 
+  public void grant(String name, Principal p, SqlAccessType grant, boolean caseSensitive) {
+    for (Map.Entry<String, Map<Principal, SqlAccessType>> entry
+        : privilegeMap.range(name, caseSensitive).entrySet()) {
+      final Map<Principal, SqlAccessType> tablePrivileges = entry.getValue();
+      final SqlAccessType accessType =
+          tablePrivileges.getOrDefault(p, SqlAccessType.createNoneAccess());
+      accessType.add(grant);
+      tablePrivileges.put(p, accessType);
+      return;
+    }
+
+    final Map<Principal, SqlAccessType> tablePrivileges = new HashMap<>();
+    tablePrivileges.put(p, grant);
+    privilegeMap.put(name, tablePrivileges);
+  }
+
+  public void revoke(String name, Principal p, SqlAccessType revoke, boolean caseSensitive) {
+    for (Map.Entry<String, Map<Principal, SqlAccessType>> entry
+        : privilegeMap.range(name, caseSensitive).entrySet()) {
+      final Map<Principal, SqlAccessType> tablePrivileges = entry.getValue();
+      final SqlAccessType accessType = tablePrivileges.get(p);
+      if (accessType == null) {
+        return;
+      }
+      accessType.remove(revoke);
+      return;
+    }
+  }
+
+  public SqlAccessType getAccessType(String name, Principal p, boolean caseSensitive) {
+    for (Map.Entry<String, Map<Principal, SqlAccessType>> entry
+        : privilegeMap.range(name, caseSensitive).entrySet()) {
+      final Map<Principal, SqlAccessType> tablePrivileges = entry.getValue();
+      final SqlAccessType accessType = tablePrivileges.get(p);
+      if (accessType != null) {
+        return accessType;
+      }
+    }
+    return SqlAccessType.createNoneAccess();
+  }
+
   /**
    * Entry in a schema, such as a table or sub-schema.
    *