Introduce SqlSettings for handling per-client configuration

To avoid leaking client information across the entire code-base, client
settings like TimeZone or pagination are stored in
SqlSession>SqlSettings which are available as a ThreadLocal (during
analysis) so that components that need them, can pick them up.

Since ES internally uses Joda, the date/time functionality relies on Joda,
whenever possible to match the behavior.

Original commit: elastic/x-pack-elasticsearch@20f41e2bb3

sql/jdbc-proto/src/main/java/org/elasticsearch/xpack/sql/jdbc/net/protocol/QueryInitRequest.java

@@ -15,29 +15,29 @@ import java.util.Objects;
 import java.util.TimeZone;
 
 public class QueryInitRequest extends Request {
-    public final int fetchSize;
     public final String query;
+    public final int fetchSize;
     public final TimeZone timeZone;
     public final TimeoutInfo timeout;
 
-    public QueryInitRequest(int fetchSize, String query, TimeZone timeZone, TimeoutInfo timeout) {
+    public QueryInitRequest(String query, int fetchSize, TimeZone timeZone, TimeoutInfo timeout) {
-        this.fetchSize = fetchSize;
         this.query = query;
+        this.fetchSize = fetchSize;
         this.timeZone = timeZone;
         this.timeout = timeout;
     }
 
     QueryInitRequest(int clientVersion, DataInput in) throws IOException {
-        fetchSize = in.readInt();
         query = in.readUTF();
+        fetchSize = in.readInt();
         timeZone = TimeZone.getTimeZone(in.readUTF());
         timeout = new TimeoutInfo(in);
     }
 
     @Override
     public void write(DataOutput out) throws IOException {
-        out.writeInt(fetchSize);
         out.writeUTF(query);
+        out.writeInt(fetchSize);
         out.writeUTF(timeZone.getID());
         timeout.write(out);
     }

sql/jdbc-proto/src/test/java/org/elasticsearch/xpack/sql/jdbc/net/protocol/QueryInitRequestTests.java

@@ -16,7 +16,7 @@ import static org.elasticsearch.xpack.sql.jdbc.net.protocol.TimeoutInfoTests.ran
 
 public class QueryInitRequestTests extends ESTestCase {
     public static QueryInitRequest randomQueryInitRequest() {
-        return new QueryInitRequest(between(0, Integer.MAX_VALUE), randomAlphaOfLength(5), randomTimeZone(random()), randomTimeoutInfo());
+        return new QueryInitRequest(randomAlphaOfLength(5), between(0, Integer.MAX_VALUE), randomTimeZone(random()), randomTimeoutInfo());
     }
 
     public void testRoundTrip() throws IOException {
@@ -25,8 +25,8 @@ public class QueryInitRequestTests extends ESTestCase {
 
     public void testToString() {
         assertEquals("QueryInitRequest<query=[SELECT * FROM test.doc]>",
-                new QueryInitRequest(10, "SELECT * FROM test.doc", TimeZone.getTimeZone("UTC"), new TimeoutInfo(1, 1, 1)).toString());
+                new QueryInitRequest("SELECT * FROM test.doc", 10, TimeZone.getTimeZone("UTC"), new TimeoutInfo(1, 1, 1)).toString());
         assertEquals("QueryInitRequest<query=[SELECT * FROM test.doc] timeZone=[GMT-05:00]>",
-                new QueryInitRequest(10, "SELECT * FROM test.doc", TimeZone.getTimeZone("GMT-5"), new TimeoutInfo(1, 1, 1)).toString());
+                new QueryInitRequest("SELECT * FROM test.doc", 10, TimeZone.getTimeZone("GMT-5"), new TimeoutInfo(1, 1, 1)).toString());
     }
 }

sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcConfiguration.java

@@ -44,8 +44,13 @@ public class JdbcConfiguration extends ConnectionConfiguration {
     // can be out/err/url
     static final String DEBUG_OUTPUT_DEFAULT = "err";
 
-    static final String TIME_ZONE = "time_zone";
+    static final String TIME_ZONE = "timezone";
-    static final String TIME_ZONE_DEFAULT = "UTC_CALENDAR";
+
+    // follow the JDBC spec and use the JVM default...
+    // to avoid inconsistency, the default is picked up once at startup and reused across connections
+    // to cater to the principle of least surprise 
+    // really, the way to move forward is to specify a calendar or the timezone manually
+    static final String TIME_ZONE_DEFAULT = TimeZone.getDefault().getID();
 
     private static final List<String> KNOWN_OPTIONS = Arrays.asList(DEBUG, DEBUG_OUTPUT, TIME_ZONE);
 

sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcDriver.java

@@ -5,6 +5,9 @@
  */
 package org.elasticsearch.xpack.sql.jdbc.jdbc;
 
+import org.elasticsearch.xpack.sql.jdbc.debug.Debug;
+import org.elasticsearch.xpack.sql.jdbc.util.Version;
+
 import java.io.Closeable;
 import java.sql.Connection;
 import java.sql.DriverManager;
@@ -15,9 +18,6 @@ import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Logger;
 
-import org.elasticsearch.xpack.sql.jdbc.debug.Debug;
-import org.elasticsearch.xpack.sql.jdbc.util.Version;
-
 public class JdbcDriver implements java.sql.Driver, Closeable {
 
     static {
@@ -55,7 +55,7 @@ public class JdbcDriver implements java.sql.Driver, Closeable {
     private static JdbcConfiguration initInfo(String url, Properties props) {
         JdbcConfiguration ci = new JdbcConfiguration(url, props);
         if (DriverManager.getLoginTimeout() > 0) {
-            ci.setConnectTimeout(TimeUnit.SECONDS.toMillis(DriverManager.getLoginTimeout()));
+            ci.connectTimeout(TimeUnit.SECONDS.toMillis(DriverManager.getLoginTimeout()));
         }
         return ci;
     }

sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcResultSet.java

@@ -40,10 +40,10 @@ import static java.lang.String.format;
 class JdbcResultSet implements ResultSet, JdbcWrapper {
 
     // temporary calendar instance (per connection) used for normalizing the date and time
-    // even though the info is already in UTC_CALENDAR format, JDBC 3.0 requires java.sql.Time to have its date
+    // even though the cfg is already in UTC format, JDBC 3.0 requires java.sql.Time to have its date
     // removed (set to Jan 01 1970) and java.sql.Date to have its HH:mm:ss component removed
     // instead of dealing with longs, a Calendar object is used instead
-    private final Calendar DEFAULT_CALENDAR = TypeConverter.defaultCalendar();
+    private final Calendar defaultCalendar;
 
     private final JdbcStatement statement;
     private final Cursor cursor;
@@ -57,6 +57,8 @@ class JdbcResultSet implements ResultSet, JdbcWrapper {
     JdbcResultSet(JdbcStatement statement, Cursor cursor) {
         this.statement = statement;
         this.cursor = cursor;
+        // TODO: should we consider the locale as well? 
+        this.defaultCalendar = Calendar.getInstance(statement.cfg.timeZone(), Locale.ROOT);
 
         List<ColumnInfo> columns = cursor.columns();
         for (int i = 0; i < columns.size(); i++) {
@@ -239,7 +241,7 @@ class JdbcResultSet implements ResultSet, JdbcWrapper {
     }
 
     private Calendar safeCalendar(Calendar calendar) {
-        return calendar == null ? DEFAULT_CALENDAR : calendar;
+        return calendar == null ? defaultCalendar : calendar;
     }
 
     @Override

sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcStatement.java

@@ -5,6 +5,9 @@
  */
 package org.elasticsearch.xpack.sql.jdbc.jdbc;
 
+import org.elasticsearch.xpack.sql.jdbc.net.client.Cursor;
+import org.elasticsearch.xpack.sql.jdbc.net.client.RequestMeta;
+
 import java.sql.Connection;
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -13,13 +16,10 @@ import java.sql.SQLWarning;
 import java.sql.Statement;
 import java.util.concurrent.TimeUnit;
 
-import org.elasticsearch.xpack.sql.jdbc.net.client.Cursor;
-import org.elasticsearch.xpack.sql.jdbc.net.client.RequestMeta;
-
 class JdbcStatement implements Statement, JdbcWrapper {
 
     final JdbcConnection con;
-    final JdbcConfiguration info;
+    final JdbcConfiguration cfg;
 
     private boolean closed = false;
     private boolean closeOnCompletion = false;
@@ -30,7 +30,7 @@ class JdbcStatement implements Statement, JdbcWrapper {
 
     JdbcStatement(JdbcConnection jdbcConnection, JdbcConfiguration info) {
         this.con = jdbcConnection;
-        this.info = info;
+        this.cfg = info;
     }
 
     @Override
@@ -155,7 +155,7 @@ class JdbcStatement implements Statement, JdbcWrapper {
         // close previous result set
         closeResultSet();
 
-        Cursor cursor = con.client.query(sql, info.timeZone(), requestMeta);
+        Cursor cursor = con.client.query(sql, requestMeta);
         rs = new JdbcResultSet(this, cursor);
     }
 

sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/TypeConverter.java

@@ -18,7 +18,6 @@ import java.time.OffsetTime;
 import java.util.Calendar;
 import java.util.GregorianCalendar;
 import java.util.Locale;
-import java.util.TimeZone;
 import java.util.function.Function;
 
 import static java.lang.String.format;
@@ -33,14 +32,8 @@ import static java.util.Calendar.YEAR;
 
 abstract class TypeConverter {
 
-    static final Calendar UTC_CALENDAR = Calendar.getInstance(TimeZone.getTimeZone("UTC_CALENDAR"), Locale.ROOT);
-
     private static final long DAY_IN_MILLIS = 60 * 60 * 24;
 
-    static Calendar defaultCalendar() {
-        return (Calendar) UTC_CALENDAR.clone();
-    }
-
     static Date convertDate(Long millis, Calendar cal) {
         return dateTimeConvert(millis, cal, c -> {
             c.set(HOUR_OF_DAY, 0);

sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbcx/JdbcDataSource.java

@@ -5,6 +5,10 @@
  */
 package org.elasticsearch.xpack.sql.jdbc.jdbcx;
 
+import org.elasticsearch.xpack.sql.jdbc.debug.Debug;
+import org.elasticsearch.xpack.sql.jdbc.jdbc.JdbcConfiguration;
+import org.elasticsearch.xpack.sql.jdbc.jdbc.JdbcConnection;
+
 import java.io.Closeable;
 import java.io.PrintWriter;
 import java.sql.Connection;
@@ -17,10 +21,6 @@ import java.util.logging.Logger;
 
 import javax.sql.DataSource;
 
-import org.elasticsearch.xpack.sql.jdbc.debug.Debug;
-import org.elasticsearch.xpack.sql.jdbc.jdbc.JdbcConfiguration;
-import org.elasticsearch.xpack.sql.jdbc.jdbc.JdbcConnection;
-
 public class JdbcDataSource implements DataSource, Wrapper, Closeable {
 
     private String url;
@@ -86,7 +86,7 @@ public class JdbcDataSource implements DataSource, Wrapper, Closeable {
     private Connection doGetConnection(Properties p) {
         JdbcConfiguration ci = new JdbcConfiguration(url, p);
         if (loginTimeout > 0) {
-            ci.setConnectTimeout(TimeUnit.SECONDS.toMillis(loginTimeout));
+            ci.connectTimeout(TimeUnit.SECONDS.toMillis(loginTimeout));
         }
         return new JdbcConnection(ci);
     }

sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/net/client/HttpClient.java

@@ -33,11 +33,11 @@ class HttpClient {
     }
 
     void setNetworkTimeout(long millis) {
-        cfg.setNetworkTimeout(millis);
+        cfg.networkTimeout(millis);
     }
 
     long getNetworkTimeout() {
-        return cfg.getNetworkTimeout();
+        return cfg.networkTimeout();
     }
 
     private URL url(String subPath) {

sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/net/client/JdbcHttpClient.java

@@ -37,7 +37,6 @@ import java.io.IOException;
 import java.sql.SQLException;
 import java.time.Instant;
 import java.util.List;
-import java.util.TimeZone;
 
 public class JdbcHttpClient implements Closeable {
     @FunctionalInterface
@@ -65,9 +64,9 @@ public class JdbcHttpClient implements Closeable {
         }
     }
 
-    public Cursor query(String sql, TimeZone timeZone, RequestMeta meta) throws SQLException {
+    public Cursor query(String sql, RequestMeta meta) throws SQLException {
         int fetch = meta.fetchSize() >= 0 ? meta.fetchSize() : conCfg.pageSize();
-        QueryInitRequest request = new QueryInitRequest(fetch, sql, timeZone, timeout(meta));
+        QueryInitRequest request = new QueryInitRequest(sql, fetch, conCfg.timeZone(), timeout(meta));
         BytesArray ba = http.put(out -> Proto.INSTANCE.writeRequest(request, out));
         QueryInitResponse response = doIO(ba, in -> (QueryInitResponse) readResponse(request, in));
         return new DefaultCursor(this, response.requestId, (Page) response.data, meta);
@@ -149,8 +148,8 @@ public class JdbcHttpClient implements Closeable {
         // timeout (in ms)
         long timeout = meta.timeoutInMs();
         if (timeout == 0) {
-            timeout = conCfg.getQueryTimeout();
+            timeout = conCfg.queryTimeout();
         }
-        return new TimeoutInfo(clientTime, timeout, conCfg.getPageTimeout());
+        return new TimeoutInfo(clientTime, timeout, conCfg.pageTimeout());
     }
 }

sql/net-client/src/main/java/org/elasticsearch/xpack/sql/net/client/ConnectionConfiguration.java

@@ -108,31 +108,31 @@ public class ConnectionConfiguration {
         return ssl;
     }
 
-    public void setConnectTimeout(long millis) {
+    public void connectTimeout(long millis) {
         connectTimeout = millis;
     }
 
-    public long getConnectTimeout() {
+    public long connectTimeout() {
         return connectTimeout;
     }
 
-    public void setNetworkTimeout(long millis) {
+    public void networkTimeout(long millis) {
         networkTimeout = millis;
     }
 
-    public long getNetworkTimeout() {
+    public long networkTimeout() {
         return networkTimeout;
     }
 
-    public void setQueryTimeout(long millis) {
+    public void queryTimeout(long millis) {
         queryTimeout = millis;
     }
 
-    public long getQueryTimeout() {
+    public long queryTimeout() {
         return queryTimeout;
     }
 
-    public long getPageTimeout() {
+    public long pageTimeout() {
         return pageTimeout;
     }
 

sql/net-client/src/main/java/org/elasticsearch/xpack/sql/net/client/JreHttpUrlConnection.java

@@ -40,8 +40,8 @@ public class JreHttpUrlConnection implements Closeable {
             throw new ClientException(ex, "Cannot setup connection to %s (%s)", url, ex.getMessage());
         }
 
-        con.setConnectTimeout((int) cfg.getConnectTimeout());
+        con.setConnectTimeout((int) cfg.connectTimeout());
-        con.setReadTimeout((int) cfg.getNetworkTimeout());
+        con.setReadTimeout((int) cfg.networkTimeout());
 
         con.setAllowUserInteraction(false);
         con.setUseCaches(false);

sql/server/src/main/java/org/elasticsearch/xpack/sql/analysis/analyzer/Analyzer.java

@@ -47,6 +47,7 @@ import org.elasticsearch.xpack.sql.plan.logical.UnresolvedRelation;
 import org.elasticsearch.xpack.sql.plan.logical.With;
 import org.elasticsearch.xpack.sql.rule.Rule;
 import org.elasticsearch.xpack.sql.rule.RuleExecutor;
+import org.elasticsearch.xpack.sql.session.SqlSession;
 import org.elasticsearch.xpack.sql.tree.Node;
 import org.elasticsearch.xpack.sql.type.CompoundDataType;
 import org.elasticsearch.xpack.sql.util.StringUtils;
@@ -641,8 +642,7 @@ public class Analyzer extends RuleExecutor<LogicalPlan> {
                         // TODO: might be removed
                         // dedicated count optimization
                         if (name.toUpperCase(Locale.ROOT).equals("COUNT")) {
-                            uf = new UnresolvedFunction(uf.location(), uf.name(), uf.distinct(), uf.timeZone(),
+                            uf = new UnresolvedFunction(uf.location(), uf.name(), uf.distinct(), singletonList(Literal.of(uf.arguments().get(0).location(), Integer.valueOf(1))));
-                                    singletonList(Literal.of(uf.arguments().get(0).location(), Integer.valueOf(1))));
                         }
                     }
 
@@ -666,7 +666,7 @@ public class Analyzer extends RuleExecutor<LogicalPlan> {
                         throw new UnknownFunctionException(name, uf);
                     }
                     // TODO: look into Generator for significant terms, etc..
-                    Function f = functionRegistry.resolveFunction(uf);
+                    Function f = functionRegistry.resolveFunction(uf, SqlSession.CURRENT.get());
 
                     list.add(f);
                     return f;

sql/server/src/main/java/org/elasticsearch/xpack/sql/execution/PlanExecutor.java

@@ -22,12 +22,10 @@ import org.elasticsearch.xpack.sql.session.SqlSession;
 import org.elasticsearch.xpack.sql.session.SqlSettings;
 
 import java.io.IOException;
-import java.util.TimeZone;
 import java.util.function.Supplier;
 
 public class PlanExecutor extends AbstractLifecycleComponent {
     // NOCOMMIT prefer not to use AbstractLifecycleComponent because the reasons for its tradeoffs is lost to the mists of time
-    private static final SqlSettings DEFAULTS = SqlSettings.EMPTY;
 
     private final Client client;
 
@@ -55,13 +53,17 @@ public class PlanExecutor extends AbstractLifecycleComponent {
         return catalog;
     }
 
-    public SqlSession newSession() {
+    public SqlSession newSession(SqlSettings settings) {
-        return new SqlSession(DEFAULTS, client, parser, catalog, functionRegistry, analyzer, optimizer, planner);
+        return new SqlSession(settings, client, parser, catalog, functionRegistry, analyzer, optimizer, planner);
     }
 
-    public void sql(String sql, TimeZone timeZone, ActionListener<RowSetCursor> listener) {
+    public void sql(String sql, ActionListener<RowSetCursor> listener) {
-        SqlSession session = newSession();
+        sql(SqlSettings.EMPTY, sql, listener);
-        session.executable(sql, timeZone).execute(session, listener);
+    }
+
+    public void sql(SqlSettings sqlSettings, String sql, ActionListener<RowSetCursor> listener) {
+        SqlSession session = newSession(sqlSettings);
+        session.executable(sql).execute(session, listener);
     }
 
     @Override

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/AbstractFunctionRegistry.java

@@ -8,21 +8,25 @@ package org.elasticsearch.xpack.sql.expression.function;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
 import org.elasticsearch.xpack.sql.expression.Expression;
+import org.elasticsearch.xpack.sql.expression.function.aware.DistinctAware;
+import org.elasticsearch.xpack.sql.expression.function.aware.TimeZoneAware;
 import org.elasticsearch.xpack.sql.parser.ParsingException;
+import org.elasticsearch.xpack.sql.session.SqlSettings;
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.Node;
 import org.elasticsearch.xpack.sql.tree.NodeUtils;
 import org.elasticsearch.xpack.sql.tree.NodeUtils.NodeInfo;
 import org.elasticsearch.xpack.sql.util.Assert;
 import org.elasticsearch.xpack.sql.util.StringUtils;
+import org.joda.time.DateTimeZone;
 
 import java.lang.reflect.InvocationTargetException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.TimeZone;
 import java.util.regex.Pattern;
 
 import static java.util.Collections.emptyList;
@@ -50,12 +54,12 @@ abstract class AbstractFunctionRegistry implements FunctionRegistry {
 
     
     @Override
-    public Function resolveFunction(UnresolvedFunction ur) {
+    public Function resolveFunction(UnresolvedFunction ur, SqlSettings settings) {
         FunctionDefinition def = defs.get(normalize(ur.name()));
         if (def == null) {
             throw new SqlIllegalArgumentException("Cannot find function %s; this should have been caught during analysis", ur.name());
         }
-        return createInstance(def.clazz(), ur);
+        return createInstance(def.clazz(), ur, settings);
     }
 
     @Override
@@ -95,42 +99,62 @@ abstract class AbstractFunctionRegistry implements FunctionRegistry {
     }
 
     @SuppressWarnings("rawtypes")
-    private static Function createInstance(Class<? extends Function> clazz, UnresolvedFunction ur) {
+    private static Function createInstance(Class<? extends Function> clazz, UnresolvedFunction ur, SqlSettings settings) {
         NodeInfo info = NodeUtils.info((Class<? extends Node>) clazz);
         Class<?> exp = ur.children().size() == 1 ? Expression.class : List.class;
         Object expVal = exp == Expression.class ? ur.children().get(0) : ur.children();
 
-        boolean distinctAware = true;
+        boolean noExpression = false;
-        boolean noArgument = false;
+        boolean distinctAware = DistinctAware.class.isAssignableFrom(clazz);
-        boolean tzAware = false;
+        boolean timezoneAware = TimeZoneAware.class.isAssignableFrom(clazz);
-        // distinct ctor
-        if (!Arrays.equals(new Class[] { Location.class, exp, boolean.class }, info.ctr.getParameterTypes())) {
-            if (ur.distinct()) {
-                throw new ParsingException(ur.location(), "Function [%s] does not support DISTINCT yet it was specified", ur.name());
-            }
-            distinctAware = false;
 
-            // might be a constant function
+        // check constructor signature
-            if (expVal instanceof List && ((List) expVal).isEmpty()) {
+        
-                noArgument = Arrays.equals(new Class[] { Location.class }, info.ctr.getParameterTypes());
+        
-            }
+        // validate distinct ctor
-            else if (Arrays.equals(new Class[] { Location.class, exp, TimeZone.class }, info.ctr.getParameterTypes())) {
+        if (!distinctAware && ur.distinct()) {
-                tzAware = true;
+            throw new ParsingException(ur.location(), "Function [%s] does not support DISTINCT yet it was specified", ur.name());
-            }
-            // distinctless
-            else if (!Arrays.equals(new Class[] { Location.class, exp }, info.ctr.getParameterTypes())) {
-                throw new SqlIllegalArgumentException("No constructor with signature [%s, %s (,%s)?] found for [%s]",
-                        Location.class, exp, boolean.class, clazz.getTypeName());
-            }
         }
         
+        List<Class> ctorSignature = new ArrayList<>();
+        ctorSignature.add(Location.class);
+        
+        // might be a constant function
+        if (expVal instanceof List && ((List) expVal).isEmpty()) {
+            noExpression = Arrays.equals(new Class[] { Location.class }, info.ctr.getParameterTypes());
+        }
+        else {
+            ctorSignature.add(exp);
+        }
+
+        // aware stuff
+        if (distinctAware) {
+            ctorSignature.add(boolean.class);
+        }
+        if (timezoneAware) {
+            ctorSignature.add(DateTimeZone.class);
+        }
+        
+        // validate
+        Assert.isTrue(Arrays.equals(ctorSignature.toArray(new Class[ctorSignature.size()]), info.ctr.getParameterTypes()),
+                "No constructor with signature %s found for [%s]", ctorSignature, clazz.getTypeName());
+        
+        // now add the actual values
         try {
-            // NOCOMMIT reflection here feels icky
+            List<Object> args = new ArrayList<>();
-            Object[] args;
+
-            if (tzAware) {
+            // always add location first
-                args = new Object[] { ur.location(), expVal, ur.timeZone() };
+            args.add(ur.location());
-            } else {
+
-                args = noArgument ? new Object[] { ur.location() } : (distinctAware ? new Object[] { ur.location(), expVal, ur.distinct() } : new Object[] { ur.location(), expVal });
+            // has multiple arguments
+            if (!noExpression) {
+                args.add(expVal);
+                if (distinctAware) {
+                    args.add(ur.distinct());
+                }
+                if (timezoneAware) {
+                    args.add(settings.timeZone());
+                }
             } 
             return (Function) info.ctr.newInstance(args);
         } catch (InstantiationException | IllegalAccessException | IllegalArgumentException | InvocationTargetException ex) {

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/FunctionRegistry.java

@@ -5,11 +5,13 @@
  */
 package org.elasticsearch.xpack.sql.expression.function;
 
+import org.elasticsearch.xpack.sql.session.SqlSettings;
+
 import java.util.Collection;
 
 public interface FunctionRegistry {
 
-    Function resolveFunction(UnresolvedFunction ur);
+    Function resolveFunction(UnresolvedFunction ur, SqlSettings settings);
 
     boolean functionExists(String name);
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/UnresolvedFunction.java

@@ -13,19 +13,16 @@ import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.type.DataType;
 
 import java.util.List;
-import java.util.TimeZone;
 
 public class UnresolvedFunction extends Function implements Unresolvable {
 
     private final String name;
     private final boolean distinct;
-    private final TimeZone timeZone;
 
-    public UnresolvedFunction(Location location, String name, boolean distinct, TimeZone timeZone, List<Expression> children) {
+    public UnresolvedFunction(Location location, String name, boolean distinct, List<Expression> children) {
         super(location, children);
         this.name = name;
         this.distinct = distinct;
-        this.timeZone = timeZone;
     }
 
     @Override
@@ -47,10 +44,6 @@ public class UnresolvedFunction extends Function implements Unresolvable {
         return distinct;
     }
 
-    public TimeZone timeZone() {
-        return timeZone;
-    }
-
     @Override
     public DataType dataType() {
         throw new UnresolvedException("dataType", this);

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/aggregate/Count.java

@@ -7,11 +7,12 @@ package org.elasticsearch.xpack.sql.expression.function.aggregate;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.expression.NamedExpression;
+import org.elasticsearch.xpack.sql.expression.function.aware.DistinctAware;
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.type.DataType;
 import org.elasticsearch.xpack.sql.type.DataTypes;
 
-public class Count extends AggregateFunction {
+public class Count extends AggregateFunction implements DistinctAware {
 
     private final boolean distinct;
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/aware/DistinctAware.java

@@ -0,0 +1,10 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.sql.expression.function.aware;
+
+public interface DistinctAware {
+
+}

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/aware/TimeZoneAware.java

@@ -0,0 +1,10 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.sql.expression.function.aware;
+
+public interface TimeZoneAware {
+
+}

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateTimeFunction.java

@@ -9,6 +9,7 @@ import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.expression.Expressions;
 import org.elasticsearch.xpack.sql.expression.FieldAttribute;
 import org.elasticsearch.xpack.sql.expression.function.aggregate.AggregateFunctionAttribute;
+import org.elasticsearch.xpack.sql.expression.function.aware.TimeZoneAware;
 import org.elasticsearch.xpack.sql.expression.function.scalar.ColumnProcessor;
 import org.elasticsearch.xpack.sql.expression.function.scalar.ScalarFunction;
 import org.elasticsearch.xpack.sql.expression.function.scalar.script.ScriptTemplate;
@@ -20,20 +21,25 @@ import org.joda.time.DateTimeZone;
 import org.joda.time.ReadableDateTime;
 
 import java.time.temporal.ChronoField;
-import java.util.TimeZone;
+import java.util.Locale;
 
+import static java.lang.String.format;
 import static org.elasticsearch.xpack.sql.expression.function.scalar.script.ParamsBuilder.paramsBuilder;
 import static org.elasticsearch.xpack.sql.expression.function.scalar.script.ScriptTemplate.formatTemplate;
 
-public abstract class DateTimeFunction extends ScalarFunction {
+public abstract class DateTimeFunction extends ScalarFunction implements TimeZoneAware {
-    private final TimeZone timeZone;
 
-    // NOCOMMIT I feel like our lives could be made a lot simpler with composition instead of inheritance here
+    private final DateTimeZone timeZone;
-    public DateTimeFunction(Location location, Expression argument, TimeZone timeZone) {
+
+    public DateTimeFunction(Location location, Expression argument, DateTimeZone timeZone) {
         super(location, argument);
         this.timeZone = timeZone;
     }
     
+    public DateTimeZone timeZone() {
+        return timeZone;
+    }
+
     @Override
     protected TypeResolution resolveType() {
         return argument().dataType().same(DataTypes.DATE) ? 
@@ -62,16 +68,16 @@ public abstract class DateTimeFunction extends ScalarFunction {
     }
 
     private String createTemplate() {
-        if (timeZone.getID().equals("UTC")) {
+        if (DateTimeZone.UTC.equals(timeZone)) {
             return formatTemplate("doc[{}].value.get" + extractFunction() + "()");
         } else {
             // NOCOMMIT ewwww
-            /* This uses the Java 9 time API because Painless doesn't whitelist creation of new
+            /* This uses the Java 8 time API because Painless doesn't whitelist creation of new
              * Joda classes. */
+
+            // ideally JodaTime should be used since that's internally used and there are subtle differences between that and the JDK API
             String asInstant = formatTemplate("Instant.ofEpochMilli(doc[{}].value.millis)");
-            String zoneId = "ZoneId.of(\"" + timeZone.toZoneId().getId() + "\"";
+            return format(Locale.ROOT, "ZonedDateTime.ofInstant(%s, ZoneId.of(\"%s\")).get(ChronoField.%s)", asInstant, timeZone.getID(), chronoField().name());
-            String asZonedDateTime = "ZonedDateTime.ofInstant(" + asInstant + ", " + zoneId + "))";
-            return asZonedDateTime + ".get(ChronoField." + chronoField().name() + ")";
         }
     }
 
@@ -87,14 +93,9 @@ public abstract class DateTimeFunction extends ScalarFunction {
             if (l instanceof Long) {
                 dt = new DateTime((Long) l, DateTimeZone.UTC);
             }
-            // but date histogram returns the keys already as DateTime on UTC
             else {
                 dt = (ReadableDateTime) l;
             }
-            if (false == timeZone.getID().equals("UTC")) {
-                // TODO probably faster to use `null` for UTC like core does
-                dt = dt.toDateTime().withZone(DateTimeZone.forTimeZone(timeZone));
-            }
             return Integer.valueOf(extract(dt));
         };
     }
@@ -104,10 +105,6 @@ public abstract class DateTimeFunction extends ScalarFunction {
         return DataTypes.INTEGER;
     }
 
-    public TimeZone timeZone() {
-        return timeZone;
-    }
-
     protected abstract int extract(ReadableDateTime dt);
 
     // used for aggregration (date histogram)

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DayOfMonth.java

@@ -7,14 +7,14 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.tree.Location;
+import org.joda.time.DateTimeZone;
 import org.joda.time.ReadableDateTime;
 
 import java.time.temporal.ChronoField;
-import java.util.TimeZone;
 
 public class DayOfMonth extends DateTimeFunction {
 
-    public DayOfMonth(Location location, Expression argument, TimeZone timeZone) {
+    public DayOfMonth(Location location, Expression argument, DateTimeZone timeZone) {
         super(location, argument, timeZone);
     }
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DayOfWeek.java

@@ -7,14 +7,14 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.tree.Location;
+import org.joda.time.DateTimeZone;
 import org.joda.time.ReadableDateTime;
 
 import java.time.temporal.ChronoField;
-import java.util.TimeZone;
 
 public class DayOfWeek extends DateTimeFunction {
 
-    public DayOfWeek(Location location, Expression argument, TimeZone timeZone) {
+    public DayOfWeek(Location location, Expression argument, DateTimeZone timeZone) {
         super(location, argument, timeZone);
     }
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DayOfYear.java

@@ -7,14 +7,14 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.tree.Location;
+import org.joda.time.DateTimeZone;
 import org.joda.time.ReadableDateTime;
 
 import java.time.temporal.ChronoField;
-import java.util.TimeZone;
 
 public class DayOfYear extends DateTimeFunction {
 
-    public DayOfYear(Location location, Expression argument, TimeZone timeZone) {
+    public DayOfYear(Location location, Expression argument, DateTimeZone timeZone) {
         super(location, argument, timeZone);
     }
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/Extract.java

@@ -6,108 +6,112 @@
 package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
+import org.elasticsearch.xpack.sql.session.SqlSession;
 import org.elasticsearch.xpack.sql.tree.Location;
-
+import org.joda.time.DateTimeZone;
-import java.util.TimeZone;
 
 public enum Extract {
 
     YEAR {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return new Year(source, argument, timeZone);
         }
     },
     MONTH {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return new MonthOfYear(source, argument, timeZone);
         }
     },
     WEEK {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return new WeekOfWeekYear(source, argument, timeZone);
         }
     },
     DAY {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return new DayOfMonth(source, argument, timeZone);
         }
     },
     DAY_OF_MONTH {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return DAY.toFunction(source, argument, timeZone);
         }
     },
     DOM {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return DAY.toFunction(source, argument, timeZone);
         }
     },
     DAY_OF_WEEK {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return new DayOfWeek(source, argument, timeZone);
         }
     },
     DOW {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return DAY_OF_WEEK.toFunction(source, argument, timeZone);
         }
     },
     DAY_OF_YEAR {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return new DayOfYear(source, argument, timeZone);
         }
     },
     DOY {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return DAY_OF_YEAR.toFunction(source, argument, timeZone);
         }
     },
     HOUR {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return new HourOfDay(source, argument, timeZone);
         }
     },
     MINUTE {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return new MinuteOfHour(source, argument, timeZone);
         }
     },
     MINUTE_OF_HOUR {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return MINUTE.toFunction(source, argument, timeZone);
         }
     },
     MINUTE_OF_DAY {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return new MinuteOfDay(source, argument, timeZone);
         }
     },
     SECOND {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return new SecondOfMinute(source, argument, timeZone);
         }
     },
     SECOND_OF_MINUTE {
         @Override
-        public DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone) {
+        public DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone) {
             return SECOND.toFunction(source, argument, timeZone);
         }
     };
 
-    public abstract DateTimeFunction toFunction(Location source, Expression argument, TimeZone timeZone);
+    public DateTimeFunction toFunction(Location source, Expression argument) {
+        return toFunction(source, argument, SqlSession.CURRENT.get().timeZone());
+    }
+
+    public abstract DateTimeFunction toFunction(Location source, Expression argument, DateTimeZone timeZone);
 }

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/HourOfDay.java

@@ -7,14 +7,14 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.tree.Location;
+import org.joda.time.DateTimeZone;
 import org.joda.time.ReadableDateTime;
 
 import java.time.temporal.ChronoField;
-import java.util.TimeZone;
 
 public class HourOfDay extends DateTimeFunction {
 
-    public HourOfDay(Location location, Expression argument, TimeZone timeZone) {
+    public HourOfDay(Location location, Expression argument, DateTimeZone timeZone) {
         super(location, argument, timeZone);
     }
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/MinuteOfDay.java

@@ -7,14 +7,14 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.tree.Location;
+import org.joda.time.DateTimeZone;
 import org.joda.time.ReadableDateTime;
 
 import java.time.temporal.ChronoField;
-import java.util.TimeZone;
 
 public class MinuteOfDay extends DateTimeFunction {
 
-    public MinuteOfDay(Location location, Expression argument, TimeZone timeZone) {
+    public MinuteOfDay(Location location, Expression argument, DateTimeZone timeZone) {
         super(location, argument, timeZone);
     }
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/MinuteOfHour.java

@@ -7,14 +7,14 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.tree.Location;
+import org.joda.time.DateTimeZone;
 import org.joda.time.ReadableDateTime;
 
 import java.time.temporal.ChronoField;
-import java.util.TimeZone;
 
 public class MinuteOfHour extends DateTimeFunction {
 
-    public MinuteOfHour(Location location, Expression argument, TimeZone timeZone) {
+    public MinuteOfHour(Location location, Expression argument, DateTimeZone timeZone) {
         super(location, argument, timeZone);
     }
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/MonthOfYear.java

@@ -7,14 +7,14 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.tree.Location;
+import org.joda.time.DateTimeZone;
 import org.joda.time.ReadableDateTime;
 
 import java.time.temporal.ChronoField;
-import java.util.TimeZone;
 
 public class MonthOfYear extends DateTimeFunction {
 
-    public MonthOfYear(Location location, Expression argument, TimeZone timeZone) {
+    public MonthOfYear(Location location, Expression argument, DateTimeZone timeZone) {
         super(location, argument, timeZone);
     }
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/SecondOfMinute.java

@@ -7,14 +7,14 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.tree.Location;
+import org.joda.time.DateTimeZone;
 import org.joda.time.ReadableDateTime;
 
 import java.time.temporal.ChronoField;
-import java.util.TimeZone;
 
 public class SecondOfMinute extends DateTimeFunction {
 
-    public SecondOfMinute(Location location, Expression argument, TimeZone timeZone) {
+    public SecondOfMinute(Location location, Expression argument, DateTimeZone timeZone) {
         super(location, argument, timeZone);
     }
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/WeekOfWeekYear.java

@@ -7,14 +7,14 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.tree.Location;
+import org.joda.time.DateTimeZone;
 import org.joda.time.ReadableDateTime;
 
 import java.time.temporal.ChronoField;
-import java.util.TimeZone;
 
 public class WeekOfWeekYear extends DateTimeFunction {
 
-    public WeekOfWeekYear(Location location, Expression argument, TimeZone timeZone) {
+    public WeekOfWeekYear(Location location, Expression argument, DateTimeZone timeZone) {
         super(location, argument, timeZone);
     }
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/Year.java

@@ -7,14 +7,14 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.tree.Location;
+import org.joda.time.DateTimeZone;
 import org.joda.time.ReadableDateTime;
 
 import java.time.temporal.ChronoField;
-import java.util.TimeZone;
 
 public class Year extends DateTimeFunction {
 
-    public Year(Location location, Expression argument, TimeZone timeZone) {
+    public Year(Location location, Expression argument, DateTimeZone timeZone) {
         super(location, argument, timeZone);
     }
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/parser/AstBuilder.java

@@ -8,12 +8,7 @@ package org.elasticsearch.xpack.sql.parser;
 import org.elasticsearch.xpack.sql.parser.SqlBaseParser.SingleStatementContext;
 import org.elasticsearch.xpack.sql.plan.logical.LogicalPlan;
 
-import java.util.TimeZone;
-
 class AstBuilder extends CommandBuilder {
-    AstBuilder(TimeZone timeZone) {
-        super(timeZone);
-    }
 
     @Override
     public LogicalPlan visitSingleStatement(SingleStatementContext ctx) {

sql/server/src/main/java/org/elasticsearch/xpack/sql/parser/CommandBuilder.java

@@ -5,9 +5,6 @@
  */
 package org.elasticsearch.xpack.sql.parser;
 
-import java.util.Locale;
-import java.util.TimeZone;
-
 import org.elasticsearch.common.Booleans;
 import org.elasticsearch.xpack.sql.parser.SqlBaseParser.DebugContext;
 import org.elasticsearch.xpack.sql.parser.SqlBaseParser.ExplainContext;
@@ -31,10 +28,9 @@ import org.elasticsearch.xpack.sql.plan.logical.command.ShowSession;
 import org.elasticsearch.xpack.sql.plan.logical.command.ShowTables;
 import org.elasticsearch.xpack.sql.tree.Location;
 
+import java.util.Locale;
+
 abstract class CommandBuilder extends LogicalPlanBuilder {
-    CommandBuilder(TimeZone timeZone) {
-        super(timeZone);
-    }
 
     @Override
     public Command visitDebug(DebugContext ctx) {

sql/server/src/main/java/org/elasticsearch/xpack/sql/parser/ExpressionBuilder.java

@@ -71,29 +71,11 @@ import org.elasticsearch.xpack.sql.type.DataTypes;
 import java.math.BigDecimal;
 import java.util.List;
 import java.util.Locale;
-import java.util.TimeZone;
 import java.util.stream.Collectors;
 
 import static java.lang.String.format;
 
 abstract class ExpressionBuilder extends IdentifierBuilder {
-    /**
-     * Time zone in which to execute the query. Used by date time
-     * functions and the rounding in date histograms.
-     */
-    private final TimeZone timeZone;
-
-    ExpressionBuilder(TimeZone timeZone) {
-        this.timeZone = timeZone;
-    }
-
-    /**
-     * Time zone in which to execute the query. Used by date time
-     * functions and the rounding in date histograms.
-     */
-    protected TimeZone timeZone() {
-        return timeZone;
-    }
 
     protected Expression expression(ParseTree ctx) {
         return typedParsing(ctx, Expression.class);
@@ -304,7 +286,7 @@ abstract class ExpressionBuilder extends IdentifierBuilder {
         if (ctx.setQuantifier() != null) {
             isDistinct = (ctx.setQuantifier().DISTINCT() != null);
         }
-        return new UnresolvedFunction(source(ctx), name, isDistinct, timeZone, expressions(ctx.expression()));
+        return new UnresolvedFunction(source(ctx), name, isDistinct, expressions(ctx.expression()));
     }
 
     @Override
@@ -317,7 +299,7 @@ abstract class ExpressionBuilder extends IdentifierBuilder {
         } catch (IllegalArgumentException ex) {
             throw new ParsingException(source, format(Locale.ROOT, "Invalid EXTRACT field %s", fieldString));
         }
-        return extract.toFunction(source, expression(ctx.valueExpression()), timeZone);
+        return extract.toFunction(source, expression(ctx.valueExpression()));
     }
 
     @Override

sql/server/src/main/java/org/elasticsearch/xpack/sql/parser/LogicalPlanBuilder.java

@@ -5,11 +5,11 @@
  */
 package org.elasticsearch.xpack.sql.parser;
 
-import java.util.LinkedHashMap;
+import org.elasticsearch.xpack.sql.expression.Expression;
-import java.util.List;
+import org.elasticsearch.xpack.sql.expression.Literal;
-import java.util.Map;
+import org.elasticsearch.xpack.sql.expression.NamedExpression;
-import java.util.TimeZone;
+import org.elasticsearch.xpack.sql.expression.Order;
-
+import org.elasticsearch.xpack.sql.expression.UnresolvedAlias;
 import org.elasticsearch.xpack.sql.parser.SqlBaseParser.AliasedQueryContext;
 import org.elasticsearch.xpack.sql.parser.SqlBaseParser.AliasedRelationContext;
 import org.elasticsearch.xpack.sql.parser.SqlBaseParser.FromClauseContext;
@@ -23,17 +23,13 @@ import org.elasticsearch.xpack.sql.parser.SqlBaseParser.QuerySpecificationContex
 import org.elasticsearch.xpack.sql.parser.SqlBaseParser.RelationContext;
 import org.elasticsearch.xpack.sql.parser.SqlBaseParser.SubqueryContext;
 import org.elasticsearch.xpack.sql.parser.SqlBaseParser.TableNameContext;
-import org.elasticsearch.xpack.sql.expression.Expression;
-import org.elasticsearch.xpack.sql.expression.Literal;
-import org.elasticsearch.xpack.sql.expression.NamedExpression;
-import org.elasticsearch.xpack.sql.expression.Order;
-import org.elasticsearch.xpack.sql.expression.UnresolvedAlias;
 import org.elasticsearch.xpack.sql.plan.TableIdentifier;
 import org.elasticsearch.xpack.sql.plan.logical.Aggregate;
 import org.elasticsearch.xpack.sql.plan.logical.Distinct;
 import org.elasticsearch.xpack.sql.plan.logical.Filter;
 import org.elasticsearch.xpack.sql.plan.logical.FromlessSelect;
 import org.elasticsearch.xpack.sql.plan.logical.Join;
+import org.elasticsearch.xpack.sql.plan.logical.Join.JoinType;
 import org.elasticsearch.xpack.sql.plan.logical.Limit;
 import org.elasticsearch.xpack.sql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.sql.plan.logical.OrderBy;
@@ -41,16 +37,16 @@ import org.elasticsearch.xpack.sql.plan.logical.Project;
 import org.elasticsearch.xpack.sql.plan.logical.SubQueryAlias;
 import org.elasticsearch.xpack.sql.plan.logical.UnresolvedRelation;
 import org.elasticsearch.xpack.sql.plan.logical.With;
-import org.elasticsearch.xpack.sql.plan.logical.Join.JoinType;
 import org.elasticsearch.xpack.sql.type.DataTypes;
 
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
 import static java.util.Collections.emptyList;
 import static java.util.stream.Collectors.toList;
 
 abstract class LogicalPlanBuilder extends ExpressionBuilder {
-    LogicalPlanBuilder(TimeZone timeZone) {
-        super(timeZone);
-    }
 
     @Override
     public LogicalPlan visitQuery(QueryContext ctx) {

sql/server/src/main/java/org/elasticsearch/xpack/sql/parser/SqlParser.java

@@ -19,30 +19,29 @@ import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.plan.logical.LogicalPlan;
 
-import java.util.TimeZone;
 import java.util.function.Function;
 
 public class SqlParser {
 
     private static final Logger log = Loggers.getLogger(SqlParser.class);
 
-    public LogicalPlan createStatement(String sql, TimeZone timeZone) {
+    public LogicalPlan createStatement(String sql) {
         if (log.isDebugEnabled()) {
             log.debug("Parsing as statement: {}", sql);
         }
-        return invokeParser("statement", sql, timeZone, SqlBaseParser::singleStatement);
+        return invokeParser("statement", sql, SqlBaseParser::singleStatement);
     }
 
-    public Expression createExpression(String expression, TimeZone timeZone) {
+    public Expression createExpression(String expression) {
         if (log.isDebugEnabled()) {
             log.debug("Parsing as expression: {}", expression);
         }
 
-        return invokeParser("expression", expression, timeZone, SqlBaseParser::singleExpression);
+        return invokeParser("expression", expression, SqlBaseParser::singleExpression);
     }
 
     @SuppressWarnings("unchecked")
-    private <T> T invokeParser(String name, String sql, TimeZone timeZone, Function<SqlBaseParser, ParserRuleContext> parseFunction) {
+    private <T> T invokeParser(String name, String sql, Function<SqlBaseParser, ParserRuleContext> parseFunction) {
         try {
             SqlBaseLexer lexer = new SqlBaseLexer(new CaseInsensitiveStream(sql));
 
@@ -73,7 +72,7 @@ public class SqlParser {
 
             postProcess(lexer, parser, tree);
 
-            return (T) new AstBuilder(timeZone).visit(tree);
+            return (T) new AstBuilder().visit(tree);
         }
 
         catch (StackOverflowError e) {

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/cli/CliServer.java

@@ -23,7 +23,6 @@ import org.elasticsearch.xpack.sql.protocol.shared.Request;
 import org.elasticsearch.xpack.sql.protocol.shared.Response;
 import org.elasticsearch.xpack.sql.util.StringUtils;
 
-import java.util.TimeZone;
 import java.util.function.Supplier;
 
 import static org.elasticsearch.action.ActionListener.wrap;
@@ -78,8 +77,9 @@ public class CliServer extends AbstractSqlServer {
     public void command(CommandRequest req, ActionListener<Response> listener) {
         final long start = System.currentTimeMillis(); // NOCOMMIT should be nanoTime or else clock skew will skew us
 
+        // NOCOMMIT: need to add settings for CLI
         // TODO support non-utc for cli server
-        executor.sql(req.command, TimeZone.getTimeZone("UTC"), wrap(
+        executor.sql(req.command, wrap(
                 c -> {
                     long stop = System.currentTimeMillis();
                     String requestId = EMPTY;

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/jdbc/JdbcServer.java

@@ -9,6 +9,7 @@ import org.elasticsearch.Build;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.xpack.sql.analysis.catalog.EsType;
 import org.elasticsearch.xpack.sql.execution.PlanExecutor;
 import org.elasticsearch.xpack.sql.execution.search.SearchHitRowSetCursor;
@@ -30,6 +31,7 @@ import org.elasticsearch.xpack.sql.plugin.AbstractSqlServer;
 import org.elasticsearch.xpack.sql.protocol.shared.AbstractProto.SqlExceptionType;
 import org.elasticsearch.xpack.sql.protocol.shared.Request;
 import org.elasticsearch.xpack.sql.protocol.shared.Response;
+import org.elasticsearch.xpack.sql.session.SqlSettings;
 import org.elasticsearch.xpack.sql.type.DataType;
 import org.elasticsearch.xpack.sql.util.StringUtils;
 
@@ -141,7 +143,13 @@ public class JdbcServer extends AbstractSqlServer {
     public void queryInit(QueryInitRequest req, ActionListener<Response> listener) {
         final long start = System.currentTimeMillis();
 
-        executor.sql(req.query, req.timeZone, wrap(c -> {
+        SqlSettings sqlCfg = new SqlSettings(Settings.builder()
+                .put(SqlSettings.PAGE_SIZE, req.fetchSize)
+                .put(SqlSettings.TIMEZONE_ID, req.timeZone.getID())
+                .build()
+        );
+        
+        executor.sql(sqlCfg, req.query, wrap(c -> {
             long stop = System.currentTimeMillis();
             String requestId = EMPTY;
             if (c.hasNextSet() && c instanceof SearchHitRowSetCursor) {

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/jdbc/action/TransportJdbcAction.java

@@ -36,8 +36,7 @@ public class TransportJdbcAction extends HandledTransportAction<JdbcRequest, Jdb
         // lazy init of the resolver
         ((EsCatalog) planExecutor.catalog()).setIndexNameExpressionResolver(indexNameExpressionResolver);
         // NOCOMMIT indexNameExpressionResolver should be available some other way
-        this.jdbcServer = new JdbcServer(planExecutor, clusterService.getClusterName().value(), () -> clusterService.localNode().getName(),
+        this.jdbcServer = new JdbcServer(planExecutor, clusterService.getClusterName().value(), () -> clusterService.localNode().getName(), Version.CURRENT, Build.CURRENT);
-                Version.CURRENT, Build.CURRENT);
     }
 
     @Override

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/sql/action/SqlRequest.java

@@ -5,16 +5,16 @@
  */
 package org.elasticsearch.xpack.sql.plugin.sql.action;
 
-import java.io.IOException;
-import java.util.Objects;
-import java.util.TimeZone;
-
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.CompositeIndicesRequest;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.joda.time.DateTimeZone;
+
+import java.io.IOException;
+import java.util.Objects;
 
 import static org.elasticsearch.action.ValidateActions.addValidationError;
 
@@ -22,13 +22,13 @@ public class SqlRequest extends ActionRequest implements CompositeIndicesRequest
 
     // initialized on the first request
     private String query;
-    private TimeZone timeZone;
+    private DateTimeZone timeZone;
     // initialized after the plan has been translated
     private String sessionId;
 
     public SqlRequest() {}
 
-    public SqlRequest(String query, TimeZone timeZone, String sessionId) {
+    public SqlRequest(String query, DateTimeZone timeZone, String sessionId) {
         this.query = query;
         this.timeZone = timeZone;
         this.sessionId = sessionId;
@@ -51,7 +51,7 @@ public class SqlRequest extends ActionRequest implements CompositeIndicesRequest
         return sessionId;
     }
 
-    public TimeZone timeZone() {
+    public DateTimeZone timeZone() {
         return timeZone;
     }
 
@@ -69,7 +69,7 @@ public class SqlRequest extends ActionRequest implements CompositeIndicesRequest
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         query = in.readString();
-        timeZone = TimeZone.getTimeZone(in.readString());
+        timeZone = DateTimeZone.forID(in.readString());
         sessionId = in.readOptionalString();
     }
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/sql/action/SqlRequestBuilder.java

@@ -7,8 +7,7 @@ package org.elasticsearch.xpack.sql.plugin.sql.action;
 
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.client.ElasticsearchClient;
-
+import org.joda.time.DateTimeZone;
-import java.util.TimeZone;
 
 public class SqlRequestBuilder extends ActionRequestBuilder<SqlRequest, SqlResponse, SqlRequestBuilder> {
 
@@ -16,7 +15,7 @@ public class SqlRequestBuilder extends ActionRequestBuilder<SqlRequest, SqlRespo
         this(client, action, null, null, null);
     }
 
-    public SqlRequestBuilder(ElasticsearchClient client, SqlAction action, String query, TimeZone timeZone, String sessionId) {
+    public SqlRequestBuilder(ElasticsearchClient client, SqlAction action, String query, DateTimeZone timeZone, String sessionId) {
         super(client, action, new SqlRequest(query, timeZone, sessionId));
     }
 

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/sql/action/TransportSqlAction.java

@@ -23,8 +23,9 @@ import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
 import org.elasticsearch.xpack.sql.analysis.catalog.EsCatalog;
 import org.elasticsearch.xpack.sql.execution.PlanExecutor;
 import org.elasticsearch.xpack.sql.session.RowSetCursor;
+import org.elasticsearch.xpack.sql.session.SqlSettings;
+import org.joda.time.DateTimeZone;
 
-import java.util.TimeZone;
 import java.util.function.Supplier;
 
 import static org.elasticsearch.xpack.sql.util.ActionUtils.chain;
@@ -60,7 +61,13 @@ public class TransportSqlAction extends HandledTransportAction<SqlRequest, SqlRe
     protected void doExecute(SqlRequest request, ActionListener<SqlResponse> listener) {
         String sessionId = request.sessionId();
         String query = request.query();
-        TimeZone timeZone = request.timeZone();
+        DateTimeZone timeZone = request.timeZone();
+        
+        SqlSettings sqlCfg = new SqlSettings(
+                Settings.builder()
+                //    .put(SqlSettings.PAGE_SIZE, req.fetchSize)
+                        .put(SqlSettings.TIMEZONE_ID, request.timeZone().getID())
+                .build());
 
         try {
             if (sessionId == null) {
@@ -69,7 +76,7 @@ public class TransportSqlAction extends HandledTransportAction<SqlRequest, SqlRe
                     return;
                 }
 
-                planExecutor.sql(query, timeZone, chain(listener, c -> {
+                planExecutor.sql(query, chain(listener, c -> {
                             String id = generateId();
                             SESSIONS.put(id, c);
                             return new SqlResponse(id, c);

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/sql/rest/RestSqlAction.java

@@ -5,9 +5,6 @@
  */
 package org.elasticsearch.xpack.sql.plugin.sql.rest;
 
-import java.io.IOException;
-import java.util.TimeZone;
-
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.ParseField;
@@ -21,6 +18,9 @@ import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.xpack.sql.plugin.sql.action.SqlAction;
 import org.elasticsearch.xpack.sql.plugin.sql.action.SqlRequest;
+import org.joda.time.DateTimeZone;
+
+import java.io.IOException;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
@@ -73,7 +73,7 @@ public class RestSqlAction extends BaseRestHandler {
         }
 
         String query;
-        TimeZone timeZone;
+        DateTimeZone timeZone;
 
         static Payload from(RestRequest request) throws IOException {
             Payload payload = new Payload();
@@ -89,7 +89,7 @@ public class RestSqlAction extends BaseRestHandler {
         }
 
         public void setTimeZone(String timeZone) {
-            this.timeZone = TimeZone.getTimeZone(timeZone);
+            this.timeZone = DateTimeZone.forID(timeZone);
         }
     }
 }

sql/server/src/main/java/org/elasticsearch/xpack/sql/querydsl/agg/GroupByDateAgg.java

@@ -16,7 +16,6 @@ import org.joda.time.DateTimeZone;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.TimeZone;
 
 import static java.util.Collections.emptyList;
 import static java.util.Collections.emptyMap;
@@ -25,14 +24,14 @@ import static org.elasticsearch.search.aggregations.AggregationBuilders.dateHist
 public class GroupByDateAgg extends GroupingAgg {
 
     private final String interval;
-    private final TimeZone timeZone;
+    private final DateTimeZone timeZone;
     
-    public GroupByDateAgg(String id, String propertyPath, String fieldName, String interval, TimeZone timeZone) {
+    public GroupByDateAgg(String id, String propertyPath, String fieldName, String interval, DateTimeZone timeZone) {
         this(id, propertyPath, fieldName, interval, timeZone, emptyList(), emptyList(), emptyMap());
     }
 
-    public GroupByDateAgg(String id, String propertyPath, String fieldName, String interval, TimeZone timeZone, List<LeafAgg> subAggs,
+    public GroupByDateAgg(String id, String propertyPath, String fieldName, String interval, DateTimeZone timeZone, 
-            List<PipelineAgg> subPipelines, Map<String, Direction> order) {
+            List<LeafAgg> subAggs, List<PipelineAgg> subPipelines, Map<String, Direction> order) {
         super(id, propertyPath, fieldName, subAggs, subPipelines, order);
         this.interval = interval;
         this.timeZone = timeZone;
@@ -46,7 +45,7 @@ public class GroupByDateAgg extends GroupingAgg {
     protected AggregationBuilder toGroupingAgg() {
         DateHistogramAggregationBuilder dhab = dateHistogram(id())
                 .field(fieldName())
-                .timeZone(DateTimeZone.forTimeZone(timeZone))
+                .timeZone(timeZone)
                 .dateHistogramInterval(new DateHistogramInterval(interval));
         if (!order().isEmpty()) {
             for (Entry<String, Sort.Direction> entry : order().entrySet()) {

sql/server/src/main/java/org/elasticsearch/xpack/sql/session/SqlSession.java

@@ -18,7 +18,6 @@ import org.elasticsearch.xpack.sql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.sql.plan.physical.PhysicalPlan;
 import org.elasticsearch.xpack.sql.planner.Planner;
 
-import java.util.TimeZone;
 import java.util.function.Function;
 
 public class SqlSession {
@@ -35,6 +34,15 @@ public class SqlSession {
     private final SqlSettings defaults;
     private SqlSettings settings;
 
+    // thread-local used for sharing settings across the plan compilation
+    public static final ThreadLocal<SqlSettings> CURRENT = new ThreadLocal<SqlSettings>() {
+
+        @Override
+        public String toString() {
+            return "SQL Session";
+        }
+    };
+
     public SqlSession(SqlSession other) {
         this(other.defaults(), other.client(), other.parser, other.catalog(), other.functionRegistry(), other.analyzer(), other.optimizer(), other.planner());
     }
@@ -79,12 +87,12 @@ public class SqlSession {
         return optimizer;
     }
 
-    public LogicalPlan parse(String sql, TimeZone timeZone) {
+    public LogicalPlan parse(String sql) {
-        return parser.createStatement(sql, timeZone);
+        return parser.createStatement(sql);
     }
 
-    public Expression expression(String expression, TimeZone timeZone) {
+    public Expression expression(String expression) {
-        return parser.createExpression(expression, timeZone);
+        return parser.createExpression(expression);
     }
 
     public LogicalPlan analyzedPlan(LogicalPlan plan, boolean verify) {
@@ -99,8 +107,17 @@ public class SqlSession {
         return planner.plan(optimizedPlan(optimized), verify);
     }
 
-    public PhysicalPlan executable(String sql, TimeZone timeZone) {
+    public PhysicalPlan executable(String sql) {
-        return physicalPlan(parse(sql, timeZone), true);
+        CURRENT.set(settings);
+        try {
+            return physicalPlan(parse(sql), true);
+        } finally {
+            CURRENT.remove();
+        }
+    }
+
+    public void sql(String sql, ActionListener<RowSetCursor> listener) {
+        executable(sql).execute(this, listener);
     }
 
     public SqlSettings defaults() {

sql/server/src/main/java/org/elasticsearch/xpack/sql/session/SqlSettings.java

@@ -6,12 +6,19 @@
 package org.elasticsearch.xpack.sql.session;
 
 import org.elasticsearch.common.settings.Settings;
+import org.joda.time.DateTimeZone;
 
 // Typed object holding properties for a given 
 public class SqlSettings {
 
     public static final SqlSettings EMPTY = new SqlSettings(Settings.EMPTY);
 
+    public static final String TIMEZONE_ID = "sql.timeZoneId";
+    public static final String TIMEZONE_ID_DEFAULT = null;
+
+    public static final String PAGE_SIZE = "sql.fetch.size";
+    public static final int PAGE_SIZE_DEFAULT = 100;
+
     private final Settings cfg;
 
     public SqlSettings(Settings cfg) {
@@ -27,14 +34,15 @@ public class SqlSettings {
         return cfg.toDelimitedString(',');
     }
 
-    public boolean dateAsString() {
+    public String timeZoneId() {
-        return cfg.getAsBoolean(DATE_AS_STRING, false);
+        return cfg.get(TIMEZONE_ID, TIMEZONE_ID_DEFAULT);
     }
 
-    public SqlSettings dateAsString(boolean value) {
+    public DateTimeZone timeZone() {
-        return new SqlSettings(Settings.builder().put(cfg).put(DATE_AS_STRING, value).build());
+        return DateTimeZone.forID(TIMEZONE_ID);
     }
 
-
+    public int pageSize() {
-    private static final String DATE_AS_STRING = "sql.date_as_string";
+        return cfg.getAsInt(PAGE_SIZE, 100);
+    }
 }

sql/server/src/test/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DayOfYearTests.java

@@ -11,25 +11,24 @@ import org.elasticsearch.xpack.sql.type.DateType;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 
-import java.util.TimeZone;
-
 public class DayOfYearTests extends ESTestCase {
-    private static final TimeZone UTC = TimeZone.getTimeZone("UTC");
+    private static final DateTimeZone UTC = DateTimeZone.UTC;
+
     public void testAsColumnProcessor() {
         assertEquals(1, extract(dateTime(0), UTC));
-        assertEquals(1, extract(dateTime(0), TimeZone.getTimeZone("GMT+1")));
+        assertEquals(1, extract(dateTime(0), DateTimeZone.forID("GMT+1")));
-        assertEquals(365, extract(dateTime(0), TimeZone.getTimeZone("GMT-1")));
+        assertEquals(365, extract(dateTime(0), DateTimeZone.forID("GMT-1")));
     }
 
     private DateTime dateTime(long millisSinceEpoch) {
-        return new DateTime(millisSinceEpoch, DateTimeZone.forTimeZone(UTC));
+        return new DateTime(millisSinceEpoch, UTC);
     }
 
-    private Object extract(Object value, TimeZone timeZone) {
+    private Object extract(Object value, DateTimeZone timeZone) {
         return build(value, timeZone).asProcessor().apply(value);
     }
 
-    private DayOfYear build(Object value, TimeZone timeZone) {
+    private DayOfYear build(Object value, DateTimeZone timeZone) {
         return new DayOfYear(null, new Literal(null, value, DateType.DEFAULT), timeZone);
     }
 }