Repository: Azure/pykusto

Issue:
Autocomplete column names

Discussion:

Patch:
diff --git a/.gitignore b/.gitignore
index 939d9b9..c3c5d29 100644
--- a/.gitignore
+++ b/.gitignore
@@ -333,4 +333,9 @@ ASALocalRun/
 /.idea/workspace.xml
 /.idea/checkstyle-idea.xml
 /.idea/misc.xml
+
+# Python venv and installation files
 /venv
+/build
+/dist
+/pykusto.egg-info
diff --git a/README.md b/README.md
index 5e090c7..30a366b 100644
--- a/README.md
+++ b/README.md
@@ -16,29 +16,29 @@ pip install pykusto
 from datetime import timedelta
 from pykusto.client import PyKustoClient
 from pykusto.query import Query
-from pykusto.expressions import column_generator as col
 
 # Connect to cluster with AAD device authentication
+# Databases, tables, and columns are auto-retrieved
 client = PyKustoClient('https://help.kusto.windows.net')
 
 # Show databases
 print(client.show_databases())
 
 # Show tables in 'Samples' database
-print(client['Samples'].show_tables())
+print(client.Samples.show_tables())
 
 # Connect to 'StormEvents' table
-table = client['Samples']['StormEvents']
+t = client.Samples.StormEvents
 
 # Build query
 (
-    Query(table)        
-        # Access columns using 'col' global variable 
-        .project(col.StartTime, col.EndTime, col.EventType, col.Source)
+    Query(t)        
+        # Access columns using table variable 
+        .project(t.StartTime, t.EndTime, t.EventType, t.Source)
         # Specify new column name using Python keyword argument   
-        .extend(Duration=col.EndTime - col.StartTime)
+        .extend(Duration=t.EndTime - t.StartTime)
         # Python types are implicitly converted to Kusto types
-        .where(col.Duration > timedelta(hours=1))
+        .where(t.Duration > timedelta(hours=1))
         .take(5)
         # Output to pandas dataframe
         .to_dataframe()
diff --git a/pykusto/client.py b/pykusto/client.py
index 23b7266..379918d 100644
--- a/pykusto/client.py
+++ b/pykusto/client.py
@@ -1,28 +1,36 @@
-from typing import Union, List, Tuple
-# noinspection PyProtectedMember
+from collections import defaultdict
+from typing import Union, List, Tuple, Dict, Any
 from urllib.parse import urlparse
 
 # noinspection PyProtectedMember
 from azure.kusto.data._response import KustoResponseDataSet
 from azure.kusto.data.request import KustoClient, KustoConnectionStringBuilder, ClientRequestProperties
 
+from pykusto.expressions import BaseColumn, AnyTypeColumn
+from pykusto.item_fetcher import ItemFetcher
 from pykusto.kql_converters import KQL
+from pykusto.type_utils import INTERNAL_NAME_TO_TYPE, typed_column, DOT_NAME_TO_TYPE
 
 
-class PyKustoClient:
+class PyKustoClient(ItemFetcher):
     """
-    Handle to a Kusto cluster
+    Handle to a Kusto cluster.
+    Uses :class:`ItemFetcher` to fetch and cache the full cluster schema, including all databases, tables, columns and their types.
     """
     _client: KustoClient
     _cluster_name: str
 
-    def __init__(self, client_or_cluster: Union[str, KustoClient]) -> None:
+    def __init__(self, client_or_cluster: Union[str, KustoClient], fetch_by_default: bool = True) -> None:
         """
-        Create a new handle to Kusto cluster
+        Create a new handle to Kusto cluster. The value of "fetch_by_default" is used for current instance, and also passed on to database instances.
 
         :param client_or_cluster: Either a KustoClient object, or a cluster name. In case a cluster name is given,
             a KustoClient is generated with AAD device authentication
         """
+        self._set_client(client_or_cluster)
+        super().__init__(None, fetch_by_default)
+
+    def _set_client(self, client_or_cluster):
         if isinstance(client_or_cluster, KustoClient):
             self._client = client_or_cluster
             # noinspection PyProtectedMember
@@ -31,15 +39,21 @@ class PyKustoClient:
             self._client = self._get_client_for_cluster(client_or_cluster)
             self._cluster_name = client_or_cluster
 
+    def __repr__(self) -> str:
+        return f'PyKustoClient({self._cluster_name})'
+
+    def _new_item(self, name: str) -> 'Database':
+        # "fetch_by_default" set to false because often a database generated this way is not represented by an actual Kusto database
+        return Database(self, name, fetch_by_default=False)
+
+    def get_database(self, name: str) -> 'Database':
+        return self[name]
+
     def execute(self, database: str, query: KQL, properties: ClientRequestProperties = None) -> KustoResponseDataSet:
         return self._client.execute(database, query, properties)
 
     def show_databases(self) -> Tuple[str, ...]:
-        res: KustoResponseDataSet = self.execute('', KQL('.show databases'))
-        return tuple(r[0] for r in res.primary_results[0].rows)
-
-    def __getitem__(self, database_name: str) -> 'Database':
-        return Database(self, database_name)
+        return self.get_item_names()
 
     def get_cluster_name(self) -> str:
         return self._cluster_name
@@ -48,50 +62,117 @@ class PyKustoClient:
     def _get_client_for_cluster(cluster: str) -> KustoClient:
         return KustoClient(KustoConnectionStringBuilder.with_aad_device_authentication(cluster))
 
+    def _internal_get_items(self) -> Dict[str, 'Database']:
+        # Retrieves database names, table names, column names and types for all databases. A database name is required by the "execute" method, but is ignored for this query
+        res: KustoResponseDataSet = self.execute('', KQL('.show databases schema | project DatabaseName, TableName, ColumnName, ColumnType | limit 100000'))
+        database_to_table_to_columns = defaultdict(lambda: defaultdict(list))
+        for database_name, table_name, column_name, column_type in res.primary_results[0].rows:
+            if is_empty(database_name) or is_empty(table_name) or is_empty(column_name):
+                continue
+            database_to_table_to_columns[database_name][table_name].append(typed_column.registry[DOT_NAME_TO_TYPE[column_type]](column_name))
+        return {
+            # Database instances are provided with all table and column data, preventing them from generating more queries. However the "fetch_by_default" behavior is
+            # passed on to them for future actions.
+            database_name: Database(
+                self, database_name, {table_name: tuple(columns) for table_name, columns in table_to_columns.items()}, fetch_by_default=self._fetch_by_default
+            )
+            for database_name, table_to_columns in database_to_table_to_columns.items()
+        }
+
 
-class Database:
+class Database(ItemFetcher):
     """
-    Handle to a Kusto database
+    Handle to a Kusto database.
+    Uses :class:`ItemFetcher` to fetch and cache the full database schema, including all tables, columns and their types.
     """
     client: PyKustoClient
     name: str
 
-    def __init__(self, client: PyKustoClient, name: str) -> None:
+    def __init__(self, client: PyKustoClient, name: str, tables: Dict[str, Tuple[BaseColumn]] = None, fetch_by_default: bool = True) -> None:
+        """
+        Create a new handle to Kusto database. The value of "fetch_by_default" is used for current instance, and also passed on to database instances.
+
+        :param client: The associated PyKustoClient instance
+        :param name: Database name
+        :param tables: A mapping from table names to the columns of each table. If this is None and "fetch_by_default" is true then they will be fetched in the constructor.
+        """
+        super().__init__(
+            # Providing the items to ItemFetcher prevents further queries until the "refresh" method is explicitly called
+            None if tables is None else {table_name: Table(self, table_name, columns, fetch_by_default=fetch_by_default) for table_name, columns in tables.items()},
+            fetch_by_default
+        )
         self.client = client
         self.name = name
 
+    def __repr__(self) -> str:
+        return f'{self.client}.Database({self.name})'
+
+    def _new_item(self, name: str) -> 'Table':
+        # "fetch_by_default" set to false because often a table generated this way is not represented by an actual Kusto table
+        return Table(self, name, fetch_by_default=False)
+
     def execute(self, query: KQL, properties: ClientRequestProperties = None) -> KustoResponseDataSet:
         return self.client.execute(self.name, query, properties)
 
     def show_tables(self) -> Tuple[str, ...]:
-        res: KustoResponseDataSet = self.execute(KQL('.show tables'))
-        return tuple(r[0] for r in res.primary_results[0].rows)
+        return self.get_item_names()
 
     def get_tables(self, *tables: str):
-        return Table(self, tables)
-
-    def __getitem__(self, table_name: str) -> 'Table':
-        return self.get_tables(table_name)
-
-
-class Table:
+        return Table(self, tables, fetch_by_default=self._fetch_by_default)
+
+    def _internal_get_items(self) -> Dict[str, 'Table']:
+        # Retrieves table names, column names and types for this database only (the database name is added in the "execute" method)
+        res: KustoResponseDataSet = self.execute(KQL('.show database schema | project TableName, ColumnName, ColumnType | limit 10000'))
+        table_to_columns = defaultdict(list)
+        for table_name, column_name, column_type in res.primary_results[0].rows:
+            if is_empty(table_name) or is_empty(column_name):
+                continue
+            table_to_columns[table_name].append(typed_column.registry[DOT_NAME_TO_TYPE[column_type]](column_name))
+        # Table instances are provided with all column data, preventing them from generating more queries. However the "fetch_by_default" behavior is
+        # passed on to them for future actions.
+        return {table_name: Table(self, table_name, tuple(columns), fetch_by_default=self._fetch_by_default) for table_name, columns in table_to_columns.items()}
+
+
+class Table(ItemFetcher):
     """
-    Handle to a Kusto table
+    Handle to a Kusto table.
+    Uses :class:`ItemFetcher` to fetch and cache the table schema of columns and their types.
     """
     database: Database
-    tables: Union[str, List[str], Tuple[str, ...]]
+    tables: Tuple[str, ...]
 
-    def __init__(self, database: Database, tables: Union[str, List[str], Tuple[str, ...]]) -> None:
+    def __init__(self, database: Database, tables: Union[str, List[str], Tuple[str, ...]], columns: Tuple[BaseColumn, ...] = None, fetch_by_default: bool = True) -> None:
         """
-        Create a new handle to a Kusto table
+        Create a new handle to a Kusto table.
 
-        :param database: Database object
+        :param database: The associated Database instance
         :param tables: Either a single table name, or a list of tables. If more than one table is given OR the table
             name contains a wildcard, the Kusto 'union' statement will be used.
+        :param columns: Table columns. If this is None and "ItemFetcher" is true then they will be fetched in the constructor.
         """
-
+        super().__init__(
+            None if columns is None else {c.get_name(): c for c in columns},
+            fetch_by_default
+        )
         self.database = database
-        self.tables = [tables] if isinstance(tables, str) else tables
+        self.tables = (tables,) if isinstance(tables, str) else tuple(tables)
+
+    def __repr__(self) -> str:
+        return f'{self.database}.Table({self.get_table()})'
+
+    def _new_item(self, name: str) -> BaseColumn:
+        return AnyTypeColumn(name)
+
+    def __getattr__(self, name: str) -> Any:
+        """
+        Convenience function for obtaining a column using dot notation.
+        In contrast with the overridden method from the :class:`ItemFetcher` class, a new column is generated if needed, since new columns can be created on the fly in the course
+        of the query (e.g. using 'extend'), and there is no fear of undesired erroneous queries sent to Kusto.
+
+        :param name: Name of column
+        :return: The column with the given name
+        """
+        return self[name]
 
     def get_table(self) -> KQL:
         result = KQL(', '.join(self.tables))
@@ -114,12 +195,19 @@ class Table:
     def execute(self, rendered_query: KQL) -> KustoResponseDataSet:
         return self.database.execute(rendered_query)
 
-    def show_columns(self) -> Tuple[Tuple[str, str], ...]:
-        res: KustoResponseDataSet = self.execute(KQL('.show table {}'.format(self.get_table())))
-        return tuple(
-            (
-                r[0],  # Column name
-                r[1],  # Column type
-            )
-            for r in res.primary_results[0].rows
-        )
+    def show_columns(self):
+        return self.get_item_names()
+
+    def _internal_get_items(self) -> Dict[str, Any]:
+        # TODO: Handle unions
+        # Retrieves column names and types for this table only
+        res: KustoResponseDataSet = self.execute(KQL('.show table {} | project AttributeName, AttributeType | limit 10000'.format(self.get_table())))
+        return {
+            column_name: typed_column.registry[INTERNAL_NAME_TO_TYPE[column_type]](column_name)
+            for column_name, column_type in res.primary_results[0].rows
+        }
+
+
+# There has to be code somewhere that already does this, but I didn't find it
+def is_empty(s: str) -> bool:
+    return s is None or len(s.strip()) == 0
diff --git a/pykusto/expressions.py b/pykusto/expressions.py
index cb69d38..e6f79ae 100644
--- a/pykusto/expressions.py
+++ b/pykusto/expressions.py
@@ -1,12 +1,11 @@
 from datetime import datetime, timedelta
-from numbers import Number
 from typing import Any, List, Tuple, Mapping, Optional, Type
 from typing import Union
 
 from pykusto.kql_converters import KQL
-from pykusto.type_utils import plain_expression, aggregation_expression, KustoTypes, kql_converter, TypeName
+from pykusto.type_utils import plain_expression, aggregation_expression, PythonTypes, kql_converter, KustoType, typed_column
 
-ExpressionType = Union[KustoTypes, 'BaseExpression']
+ExpressionType = Union[PythonTypes, 'BaseExpression']
 StringType = Union[str, 'StringExpression']
 BooleanType = Union[bool, 'BooleanExpression']
 NumberType = Union[int, float, 'NumberExpression']
@@ -36,6 +35,8 @@ def _subexpr_to_kql(obj: ExpressionType) -> KQL:
 class BaseExpression:
     kql: KQL
 
+    # We would prefer to use 'abc' to make the class abstract, but this can be done only if there is at least one abstract method, which we don't have here.
+    # Overriding __new___ is the next best solution.
     def __new__(cls, *args, **kwargs):
         if cls is 'BaseExpression':
             raise TypeError("BaseExpression is abstract")
@@ -44,7 +45,8 @@ class BaseExpression:
     def __init__(self, kql: Union[KQL, 'BaseExpression']) -> None:
         if isinstance(kql, BaseExpression):
             self.kql = kql.kql
-        elif not isinstance(kql, str):
+            return
+        if not isinstance(kql, str):
             raise ValueError("Either expression or KQL required")
         self.kql = kql
 
@@ -75,7 +77,7 @@ class BaseExpression:
 
     @staticmethod
     def base_binary_op(
-            left: ExpressionType, operator: str, right: ExpressionType, result_type: Type[KustoTypes]
+            left: ExpressionType, operator: str, right: ExpressionType, result_type: Type[KustoType]
     ) -> 'BaseExpression':
         registrar = plain_expression
         if isinstance(left, AggregationExpression) or isinstance(right, AggregationExpression):
@@ -112,16 +114,25 @@ class BaseExpression:
     def to_string(self) -> 'StringExpression':
         return StringExpression(KQL('tostring({})'.format(self.kql)))
 
-    def assign_to(self, *columns: 'Column') -> 'AssignmentBase':
+    def assign_to_single_column(self, column: 'AnyTypeColumn') -> 'AssignmentToSingleColumn':
+        return AssignmentToSingleColumn(column, self)
+
+    def assign_to_multiple_columns(self, *columns: 'AnyTypeColumn') -> 'AssignmentBase':
+        """
+        This method exists for the sole purpose of providing an informative error message.
+        """
+        raise ValueError("Only arrays can be assigned to multiple columns")
+
+    def assign_to(self, *columns: 'AnyTypeColumn') -> 'AssignmentBase':
         if len(columns) == 0:
             # Unspecified column name
             return AssignmentBase(None, self)
         if len(columns) == 1:
-            return AssignmentToSingleColumn(columns[0], self)
-        raise ValueError("Only arrays can be assigned to multiple columns")
+            return self.assign_to_single_column(columns[0])
+        return self.assign_to_multiple_columns(*columns)
 
 
-@plain_expression(bool)
+@plain_expression(KustoType.BOOL)
 class BooleanExpression(BaseExpression):
     @staticmethod
     def binary_op(left: ExpressionType, operator: str, right: ExpressionType) -> 'BooleanExpression':
@@ -138,12 +149,12 @@ class BooleanExpression(BaseExpression):
         return BooleanExpression(KQL('not({})'.format(self.kql)))
 
 
-@plain_expression(Number)
+@plain_expression(KustoType.INT, KustoType.LONG, KustoType.REAL)
 class NumberExpression(BaseExpression):
     @staticmethod
     def binary_op(left: NumberType, operator: str, right: NumberType) -> 'NumberExpression':
         # noinspection PyTypeChecker
-        return BaseExpression.base_binary_op(left, operator, right, Number)
+        return BaseExpression.base_binary_op(left, operator, right, int)
 
     def __lt__(self, other: NumberType) -> BooleanExpression:
         return BooleanExpression.binary_op(self, ' < ', other)
@@ -225,16 +236,16 @@ class NumberExpression(BaseExpression):
         return BooleanExpression(KQL('isnan({})'.format(self.kql)))
 
     def log(self) -> 'NumberExpression':
-        return NumberExpression(KQL('log({})'.format(self)))
+        return NumberExpression(KQL('log({})'.format(self.kql)))
 
     def log10(self) -> 'NumberExpression':
-        return NumberExpression(KQL('log10({})'.format(self)))
+        return NumberExpression(KQL('log10({})'.format(self.kql)))
 
     def log2(self) -> 'NumberExpression':
-        return NumberExpression(KQL('log2({})'.format(self)))
+        return NumberExpression(KQL('log2({})'.format(self.kql)))
 
     def loggamma(self) -> 'NumberExpression':
-        return NumberExpression(KQL('loggamma({})'.format(self)))
+        return NumberExpression(KQL('loggamma({})'.format(self.kql)))
 
     def round(self, precision: NumberType = None) -> 'NumberExpression':
         return NumberExpression(KQL(
@@ -242,7 +253,7 @@ class NumberExpression(BaseExpression):
         ))
 
 
-@plain_expression(str)
+@plain_expression(KustoType.STRING)
 class StringExpression(BaseExpression):
     @staticmethod
     def binary_op(left: ExpressionType, operator: str, right: ExpressionType) -> 'StringExpression':
@@ -269,9 +280,9 @@ class StringExpression(BaseExpression):
 
     def split(self, delimiter: StringType, requested_index: NumberType = None) -> 'ArrayExpression':
         if requested_index is None:
-            return ArrayExpression(KQL('split({}, {})'.format(to_kql(self.kql), to_kql(delimiter))))
+            return ArrayExpression(KQL('split({}, {})'.format(self.kql, to_kql(delimiter))))
         return ArrayExpression(KQL('split({}, {}, {})'.format(
-            to_kql(self.kql), to_kql(delimiter), to_kql(requested_index)
+            self.kql, to_kql(delimiter), to_kql(requested_index)
         )))
 
     def equals(self, other: StringType, case_sensitive: bool = False) -> BooleanExpression:
@@ -311,7 +322,7 @@ class StringExpression(BaseExpression):
         return BooleanExpression(KQL('isutf8({})'.format(self.kql)))
 
 
-@plain_expression(datetime)
+@plain_expression(KustoType.DATETIME)
 class DatetimeExpression(BaseExpression):
     @staticmethod
     def binary_op(left: ExpressionType, operator: str, right: ExpressionType) -> 'DatetimeExpression':
@@ -422,7 +433,7 @@ class DatetimeExpression(BaseExpression):
         ))
 
 
-@plain_expression(timedelta)
+@plain_expression(KustoType.TIMESPAN)
 class TimespanExpression(BaseExpression):
     @staticmethod
     def binary_op(left: ExpressionType, operator: str, right: ExpressionType) -> 'TimespanExpression':
@@ -458,7 +469,7 @@ class TimespanExpression(BaseExpression):
         )))
 
 
-@plain_expression(List, Tuple)
+@plain_expression(KustoType.ARRAY)
 class ArrayExpression(BaseExpression):
     def __getitem__(self, index: NumberType) -> 'AnyExpression':
         return AnyExpression(KQL('{}[{}]'.format(self.kql, _subexpr_to_kql(index))))
@@ -469,7 +480,7 @@ class ArrayExpression(BaseExpression):
     def array_length(self) -> NumberExpression:
         return NumberExpression(KQL('array_length({})'.format(self.kql)))
 
-    def contains(self, other: ExpressionType) -> 'BooleanExpression':
+    def array_contains(self, other: ExpressionType) -> 'BooleanExpression':
         return BooleanExpression.binary_op(other, ' in ', self)
 
     @staticmethod
@@ -478,16 +489,14 @@ class ArrayExpression(BaseExpression):
             ', '.join('{}'.format(_subexpr_to_kql(e) for e in elements))
         )))
 
-    def assign_to(self, *columns: 'Column') -> 'AssignmentBase':
-        if len(columns) <= 1:
-            return super().assign_to(*columns)
+    def assign_to_multiple_columns(self, *columns: 'AnyTypeColumn') -> 'AssignmentToMultipleColumns':
         return AssignmentToMultipleColumns(columns, self)
 
 
-@plain_expression(Mapping)
+@plain_expression(KustoType.MAPPING)
 class MappingExpression(BaseExpression):
     def __getitem__(self, index: StringType) -> 'AnyExpression':
-        return AnyExpression(KQL('{}[{}]'.format(self.kql, _subexpr_to_kql(index))))
+        return AnyExpression(KQL('{}[{}]'.format(self.kql, to_kql(index))))
 
     def __getattr__(self, name: str) -> 'AnyExpression':
         return AnyExpression(KQL('{}.{}'.format(self.kql, name)))
@@ -516,12 +525,15 @@ class AnyExpression(
 
 
 class AggregationExpression(BaseExpression):
+
+    # We would prefer to use 'abc' to make the class abstract, but this can be done only if there is at least one abstract method, which we don't have here.
+    # Overriding __new___ is the next best solution.
     def __new__(cls, *args, **kwargs):
         if cls is 'AggregationExpression':
             raise TypeError("AggregationExpression is abstract")
         return object.__new__(cls)
 
-    def assign_to(self, *columns: 'Column') -> 'AssignmentFromAggregationToColumn':
+    def assign_to(self, *columns: 'AnyTypeColumn') -> 'AssignmentFromAggregationToColumn':
         if len(columns) == 0:
             # Unspecified column name
             return AssignmentFromAggregationToColumn(None, self)
@@ -533,37 +545,37 @@ class AggregationExpression(BaseExpression):
         return self.kql
 
 
-@aggregation_expression(bool)
+@aggregation_expression(KustoType.BOOL)
 class BooleanAggregationExpression(AggregationExpression, BooleanExpression):
     pass
 
 
-@aggregation_expression(Number)
+@aggregation_expression(KustoType.INT, KustoType.LONG, KustoType.REAL)
 class NumberAggregationExpression(AggregationExpression, NumberExpression):
     pass
 
 
-@aggregation_expression(str)
+@aggregation_expression(KustoType.STRING)
 class StringAggregationExpression(AggregationExpression, StringExpression):
     pass
 
 
-@aggregation_expression(datetime)
+@aggregation_expression(KustoType.DATETIME)
 class DatetimeAggregationExpression(AggregationExpression, DatetimeExpression):
     pass
 
 
-@aggregation_expression(timedelta)
+@aggregation_expression(KustoType.TIMESPAN)
 class TimespanAggregationExpression(AggregationExpression, TimespanExpression):
     pass
 
 
-@aggregation_expression(List, Tuple)
+@aggregation_expression(KustoType.ARRAY)
 class ArrayAggregationExpression(AggregationExpression, ArrayExpression):
     pass
 
 
-@aggregation_expression(Mapping)
+@aggregation_expression(KustoType.MAPPING)
 class MappingAggregationExpression(AggregationExpression, MappingExpression):
     pass
 
@@ -583,7 +595,7 @@ class AssignmentBase:
         return KQL('{} = {}'.format(self._lvalue, self._rvalue))
 
     @staticmethod
-    def assign(expression: ExpressionType, *columns: 'Column') -> 'AssignmentBase':
+    def assign(expression: ExpressionType, *columns: 'AnyTypeColumn') -> 'AssignmentBase':
         if len(columns) == 0:
             raise ValueError("Provide at least one column")
         if len(columns) == 1:
@@ -592,52 +604,116 @@ class AssignmentBase:
 
 
 class AssignmentToSingleColumn(AssignmentBase):
-    def __init__(self, column: 'Column', expression: ExpressionType) -> None:
+    def __init__(self, column: 'AnyTypeColumn', expression: ExpressionType) -> None:
         super().__init__(column.kql, expression)
 
 
 class AssignmentFromColumnToColumn(AssignmentToSingleColumn):
-    def __init__(self, target: 'Column', source: 'Column') -> None:
+    def __init__(self, target: 'AnyTypeColumn', source: 'BaseColumn') -> None:
         super().__init__(target, source)
 
 
 class AssignmentToMultipleColumns(AssignmentBase):
-    def __init__(self, columns: Union[List['Column'], Tuple['Column']], expression: ArrayType) -> None:
+    def __init__(self, columns: Union[List['AnyTypeColumn'], Tuple['AnyTypeColumn']], expression: ArrayType) -> None:
         super().__init__(KQL('({})'.format(', '.join(c.kql for c in columns))), expression)
 
 
 class AssignmentFromAggregationToColumn(AssignmentBase):
-    def __init__(self, column: Optional['Column'], aggregation: AggregationExpression) -> None:
+    def __init__(self, column: Optional['AnyTypeColumn'], aggregation: AggregationExpression) -> None:
         super().__init__(None if column is None else column.kql, aggregation)
 
 
-class Column(AnyExpression):
+class BaseColumn(BaseExpression):
     _name: str
 
+    # We would prefer to use 'abc' to make the class abstract, but this can be done only if there is at least one abstract method, which we don't have here.
+    # We can't define "get_kusto_type" as abstract because at least one concrete subclass (NumberColumn) does not override it.
+    # Overriding __new___ is the next best solution.
+    def __new__(cls, *args, **kwargs):
+        if cls is 'BaseColumn':
+            raise TypeError("BaseColumn is abstract")
+        return object.__new__(cls)
+
     def __init__(self, name: str) -> None:
         super().__init__(KQL("['{}']".format(name) if '.' in name else name))
         self._name = name
 
+    def get_name(self) -> str:
+        return self._name
+
     def as_subexpression(self) -> KQL:
         return self.kql
 
+    def assign_to_single_column(self, column: 'AnyTypeColumn') -> 'AssignmentFromColumnToColumn':
+        return AssignmentFromColumnToColumn(column, self)
+
+    def get_kusto_type(self) -> KustoType:
+        raise NotImplementedError("BaseColumn has no type")
+
+    def __repr__(self) -> str:
+        return f'{self.__class__.__name__}({self._name})'
+
+
+@typed_column(KustoType.INT, KustoType.LONG, KustoType.REAL)
+class NumberColumn(BaseColumn, NumberExpression):
+    pass
+
+
+@typed_column(KustoType.BOOL)
+class BooleanColumn(BaseColumn, BooleanExpression):
+    def get_kusto_type(self) -> KustoType:
+        return KustoType.BOOL
+
+
+@typed_column(KustoType.ARRAY)
+class ArrayColumn(BaseColumn, ArrayExpression):
+    def get_kusto_type(self) -> KustoType:
+        return KustoType.ARRAY
+
+
+@typed_column(KustoType.MAPPING)
+class MappingColumn(BaseColumn, MappingExpression):
+    def get_kusto_type(self) -> KustoType:
+        return KustoType.MAPPING
+
+
+class DynamicColumn(ArrayColumn, MappingColumn):
+    def get_kusto_type(self) -> KustoType:
+        raise ValueError("Column type unknown")
+
+
+@typed_column(KustoType.STRING)
+class StringColumn(BaseColumn, StringExpression):
+    def get_kusto_type(self) -> KustoType:
+        return KustoType.STRING
+
+
+@typed_column(KustoType.DATETIME)
+class DatetimeColumn(BaseColumn, DatetimeExpression):
+    def get_kusto_type(self) -> KustoType:
+        return KustoType.DATETIME
+
+
+@typed_column(KustoType.TIMESPAN)
+class TimespanColumn(BaseColumn, TimespanExpression):
+    def get_kusto_type(self) -> KustoType:
+        return KustoType.TIMESPAN
+
+
+class AnyTypeColumn(NumberColumn, BooleanColumn, DynamicColumn, StringColumn, DatetimeColumn, TimespanColumn):
     def __len__(self) -> NumberExpression:
-        raise NotImplementedError("Column type unknown, instead use 'string_size' or 'array_length'")
+        raise ValueError("Column type unknown, instead use 'string_size' or 'array_length'")
 
-    def assign_to(self, *columns: 'Column') -> 'AssignmentBase':
-        if len(columns) == 0:
-            return super().assign_to()
-        if len(columns) == 1:
-            return AssignmentFromColumnToColumn(columns[0], self)
-        return ArrayExpression.assign_to(self, *columns)
+    def get_kusto_type(self) -> KustoType:
+        raise ValueError("Column type unknown")
 
 
 class ColumnGenerator:
-    def __getattr__(self, name: str) -> Column:
-        return Column(name)
+    def __getattr__(self, name: str) -> AnyTypeColumn:
+        return AnyTypeColumn(name)
 
-    def __getitem__(self, name: str) -> Column:
-        return Column(name)
+    def __getitem__(self, name: str) -> AnyTypeColumn:
+        return AnyTypeColumn(name)
 
 
 # Recommended usage: from pykusto.expressions import column_generator as col
@@ -646,8 +722,8 @@ column_generator = ColumnGenerator()
 
 
 class ColumnToType(BaseExpression):
-    def __init__(self, col: Column, type_name: TypeName) -> None:
-        super().__init__(KQL("{} to typeof({})".format(col.kql, type_name.value)))
+    def __init__(self, col: BaseColumn, kusto_type: KustoType) -> None:
+        super().__init__(KQL("{} to typeof({})".format(col.kql, kusto_type.primary_name)))
 
 
 def to_kql(obj: ExpressionType) -> KQL:
diff --git a/pykusto/functions.py b/pykusto/functions.py
index fa01772..e55c5f7 100644
--- a/pykusto/functions.py
+++ b/pykusto/functions.py
@@ -1,13 +1,13 @@
 import json
 from typing import Union
 
-from pykusto.expressions import Column, NumberType, NumberExpression, TimespanType, \
+from pykusto.expressions import AnyTypeColumn, NumberType, NumberExpression, TimespanType, \
     DatetimeExpression, TimespanExpression, ArrayType, DynamicType, DatetimeType, BaseExpression, BooleanType, \
     ExpressionType, AggregationExpression, StringType, StringExpression, BooleanExpression, \
     NumberAggregationExpression, MappingAggregationExpression, ArrayAggregationExpression, to_kql, DynamicExpression, \
-    ArrayExpression, ColumnToType
+    ArrayExpression, ColumnToType, BaseColumn
 from pykusto.kql_converters import KQL
-from pykusto.type_utils import plain_expression, get_base_types, TypeName
+from pykusto.type_utils import plain_expression, get_base_types, KustoType
 
 
 # Scalar functions
@@ -26,7 +26,7 @@ def ago(expr: TimespanType) -> DatetimeExpression:
 
 
 def array_length(expr: ArrayType) -> NumberExpression:
-    return expr.array_length()
+    return ArrayExpression(expr).array_length()
 
 
 # def array_slice(): return
@@ -272,10 +272,10 @@ def iff(predicate: BooleanType, if_true: ExpressionType, if_false: ExpressionTyp
     if len(common_types) == 0:
         # If there is not at least one common type, then certainly the arguments are not of the same type
         raise TypeError("The second and third arguments must be of the same type, but they are: {} and {}".format(
-            ", ".join(t.__name__ for t in return_types),
-            ", ".join(t.__name__ for t in other_types)
+            ", ".join(t.primary_name for t in return_types),
+            ", ".join(t.primary_name for t in other_types)
         ))
-    return plain_expression.for_type(next(iter(common_types)))(
+    return plain_expression.registry[next(iter(common_types))](
         KQL('iff({}, {}, {})'.format(to_kql(predicate), to_kql(if_true), to_kql(if_false)))
     )
 
@@ -567,7 +567,7 @@ def sign(expr: NumberType) -> NumberExpression:
 #
 #
 def split(string: StringType, delimiter: StringType, requested_index: NumberType = None) -> 'ArrayExpression':
-    return StringExpression(KQL(string)).split(delimiter, requested_index)
+    return StringExpression(to_kql(string)).split(delimiter, requested_index)
 
 
 def sqrt(expr: NumberType) -> NumberExpression:
@@ -621,7 +621,7 @@ def strcmp(expr1: StringType, expr2: StringType) -> NumberExpression:
 
 
 def string_size(expr: StringType) -> NumberExpression:
-    return NumberExpression(KQL('string_size({})'.format(to_kql(expr))))
+    return StringExpression(expr).string_size()
 
 
 def strlen(expr: StringType) -> NumberExpression:
@@ -772,7 +772,7 @@ def avgif(expr: ExpressionType, predicate: BooleanType) -> NumberAggregationExpr
 #     return
 
 
-def count(col: Column = None) -> NumberAggregationExpression:
+def count(col: AnyTypeColumn = None) -> NumberAggregationExpression:
     res = "count()" if col is None else "count({})".format(col.kql)
     return NumberAggregationExpression(KQL(res))
 
@@ -835,7 +835,7 @@ def percentile(expr: ExpressionType, per: NumberType) -> AggregationExpression:
 
 
 def percentiles(expr: ExpressionType, *pers: NumberType) -> AggregationExpression:
-    res = 'percentiles({}, {})'.format(expr,
+    res = 'percentiles({}, {})'.format(expr.kql,
                                        ', '.join([str(to_kql(per)) for per in pers]))
     return AggregationExpression(KQL(res))
 
@@ -881,5 +881,5 @@ def variancep(expr: ExpressionType) -> AggregationExpression:
 
 
 # Used for mv-expand
-def to_type(column: Column, type_name: TypeName) -> ColumnToType:
+def to_type(column: BaseColumn, type_name: KustoType) -> ColumnToType:
     return ColumnToType(column, type_name)
diff --git a/pykusto/item_fetcher.py b/pykusto/item_fetcher.py
new file mode 100644
index 0000000..d27a855
--- /dev/null
+++ b/pykusto/item_fetcher.py
@@ -0,0 +1,114 @@
+from abc import ABCMeta, abstractmethod
+from concurrent.futures import Future, ThreadPoolExecutor, wait
+from itertools import chain
+from threading import Lock
+from typing import Union, Dict, Any, Iterable, Callable
+
+# Using a thread pool even though we only need one thread, because that's the only way to make use of "futures".
+# Also, this makes it easy to use more than one thread, if the need ever arises.
+POOL = ThreadPoolExecutor(max_workers=1)
+
+
+class ItemFetcher(metaclass=ABCMeta):
+    """
+    Abstract class that caches a collection of items, fetching them in certain scenarios.
+    """
+    _fetch_by_default: bool
+    __items: Union[None, Dict[str, Any]]
+    __future: Union[None, Future]
+    __item_write_lock: Lock
+    __item_fetch_lock: Lock
+
+    def __init__(self, items: Union[None, Dict[str, Any]], fetch_by_default: bool) -> None:
+        """
+        :param items: Initial items. If not None, items will not be fetched until the "refresh" method is explicitly called.
+        :param fetch_by_default: When true, items will be fetched in the constructor, but only if they were not supplied as a parameter. Subclasses are encouraged to pass
+                                    on the value of "fetch_by_default" to child ItemFetchers.
+        """
+        self._fetch_by_default = fetch_by_default
+        self.__items = items
+        self.__future = None
+        self.__item_write_lock = Lock()
+        self.__item_fetch_lock = Lock()
+        if items is None and fetch_by_default:
+            self.refresh()
+
+    def get_item_names(self):
+        return tuple(self.__items.keys())
+
+    @abstractmethod
+    def _new_item(self, name: str) -> Any:
+        raise NotImplementedError()
+
+    def __getattr__(self, name: str) -> Any:
+        """
+        Convenience function for obtaining an item using dot notation.
+        Often dot notation is used for other purposes, and sometimes that happens implicitly. For example Jupyter notebooks automatically run dot-notation code in the background
+        on objects. For this reason, to avoid undesired erroneous queries sent to Kusto, an item is returned only if one already exists, and a new item is not generated otherwise
+        (in contrast to bracket notation).
+
+        :param name: Name of item to return
+        :return: The item with the given name
+        :raises AttributeError: If there is no such item
+        """
+        return self._get_item(name, lambda: _raise(AttributeError(f"{self} has no attribute '{name}'")))
+
+    def __getitem__(self, name: str) -> Any:
+        """
+        Convenience function for obtaining an item using bracket notation.
+        Since bracket notation is only used explicitly, a new item is generated if needed (in contrast to dot notation).
+
+        :param name: Name of item to return
+        :return: The item with the given name
+        """
+        return self._get_item(name, lambda: self.__generate_and_save_new_item(name))
+
+    def __generate_and_save_new_item(self, name):
+        item = self._new_item(name)
+        with self.__item_write_lock:
+            if self.__items is None:
+                self.__items = {}
+            self.__items[name] = item
+        return item
+
+    def _get_item(self, name: str, fallback: Callable) -> Any:
+        if self.__items is not None:
+            resolved_item = self.__items.get(name)
+            if resolved_item is not None:
+                return resolved_item
+        return fallback()
+
+    def __dir__(self) -> Iterable[str]:
+        return sorted(chain(super().__dir__(), tuple() if self.__items is None else self.__items.keys()))
+
+    def refresh(self):
+        """
+        Fetches all items in a separate thread, making them available after the tread finishes executing. The 'wait_for_items' method can be used to wait for that to happen.
+        The specific logic for fetching is defined in concrete subclasses.
+        """
+        self.__future = POOL.submit(self._get_items)
+        self.__future.add_done_callback(self._set_items)
+
+    def wait_for_items(self):
+        """
+        If item fetching is currently in progress, wait until it is done and return, otherwise return immediately.
+        If several fetching threads are in progress, wait for the most recent one.
+        """
+        if self.__future is not None:
+            wait((self.__future,))
+
+    def _set_items(self, future: Future):
+        with self.__item_write_lock:
+            self.__items = future.result()
+
+    @abstractmethod
+    def _internal_get_items(self) -> Dict[str, Any]:
+        raise NotImplementedError()
+
+    def _get_items(self) -> Dict[str, Any]:
+        with self.__item_fetch_lock:
+            return self._internal_get_items()
+
+
+def _raise(e: BaseException):
+    raise e
diff --git a/pykusto/kql_converters.py b/pykusto/kql_converters.py
index 85c3c3a..3ee5fbf 100644
--- a/pykusto/kql_converters.py
+++ b/pykusto/kql_converters.py
@@ -4,17 +4,17 @@ from itertools import chain
 from numbers import Number
 from typing import NewType, Mapping, Union, List, Tuple
 
-from pykusto.type_utils import kql_converter
+from pykusto.type_utils import kql_converter, KustoType
 
 KQL = NewType('KQL', str)
 
 
-@kql_converter(datetime)
+@kql_converter(KustoType.DATETIME)
 def datetime_to_kql(dt: datetime) -> KQL:
     return KQL(dt.strftime('datetime(%Y-%m-%d %H:%M:%S.%f)'))
 
 
-@kql_converter(timedelta)
+@kql_converter(KustoType.TIMESPAN)
 def timedelta_to_kql(td: timedelta) -> KQL:
     hours, remainder = divmod(td.seconds, 3600)
     minutes, seconds = divmod(remainder, 60)
@@ -27,7 +27,7 @@ def timedelta_to_kql(td: timedelta) -> KQL:
     ))
 
 
-@kql_converter(Mapping, List, Tuple)
+@kql_converter(KustoType.ARRAY, KustoType.MAPPING)
 def dynamic_to_kql(d: Union[Mapping, List, Tuple]) -> KQL:
     try:
         query = list(json.dumps(d))
@@ -69,21 +69,22 @@ def build_dynamic(d: Union[Mapping, List, Tuple]) -> KQL:
     return to_kql(d)
 
 
-@kql_converter(bool)
+@kql_converter(KustoType.BOOL)
 def bool_to_kql(b: bool) -> KQL:
     return KQL('true') if b else KQL('false')
 
 
-@kql_converter(str)
+@kql_converter(KustoType.STRING)
 def str_to_kql(s: str) -> KQL:
     return KQL('"{}"'.format(s))
 
 
-@kql_converter(Number)
+@kql_converter(KustoType.INT, KustoType.LONG, KustoType.REAL)
 def number_to_kql(n: Number) -> KQL:
     return KQL(str(n))
 
 
-@kql_converter(type(None))
+# noinspection PyUnusedLocal
+@kql_converter(KustoType.NULL)
 def none_to_kql(n: type(None)) -> KQL:
     return KQL("")
diff --git a/pykusto/query.py b/pykusto/query.py
index b2f77f3..c2df0a4 100644
--- a/pykusto/query.py
+++ b/pykusto/query.py
@@ -11,11 +11,12 @@ from azure.kusto.data.helpers import dataframe_from_result_table
 
 from pykusto.client import Table
 from pykusto.expressions import BooleanType, ExpressionType, AggregationExpression, OrderedType, \
-    StringType, AssignmentBase, AssignmentFromAggregationToColumn, AssignmentToSingleColumn, Column, BaseExpression, \
+    StringType, AssignmentBase, AssignmentFromAggregationToColumn, AssignmentToSingleColumn, AnyTypeColumn, \
+    BaseExpression, \
     AssignmentFromColumnToColumn, AnyExpression, to_kql, ColumnToType
 from pykusto.kql_converters import KQL
 from pykusto.logger import logger
-from pykusto.type_utils import TypeName
+from pykusto.type_utils import KustoType
 from pykusto.udf import stringify_python_func
 
 
@@ -102,17 +103,17 @@ class Query:
     def order_by(self, col: OrderedType, order: Order = None, nulls: Nulls = None) -> 'OrderQuery':
         return OrderQuery(self, col, order, nulls)
 
-    def top(self, num_rows: int, col: Column, order: Order = None, nulls: Nulls = None) -> 'TopQuery':
+    def top(self, num_rows: int, col: AnyTypeColumn, order: Order = None, nulls: Nulls = None) -> 'TopQuery':
         return TopQuery(self, num_rows, col, order, nulls)
 
     def join(self, query: 'Query', kind: JoinKind = None) -> 'JoinQuery':
         return JoinQuery(self, query, kind)
 
-    def project(self, *args: Union[Column, AssignmentBase, BaseExpression], **kwargs: ExpressionType) -> 'ProjectQuery':
-        columns: List[Column] = []
+    def project(self, *args: Union[AnyTypeColumn, AssignmentBase, BaseExpression], **kwargs: ExpressionType) -> 'ProjectQuery':
+        columns: List[AnyTypeColumn] = []
         assignments: List[AssignmentBase] = []
         for arg in args:
-            if isinstance(arg, Column):
+            if isinstance(arg, AnyTypeColumn):
                 columns.append(arg)
             elif isinstance(arg, AssignmentBase):
                 assignments.append(arg)
@@ -121,13 +122,13 @@ class Query:
             else:
                 raise ValueError("Invalid assignment: " + arg.to_kql())
         for column_name, expression in kwargs.items():
-            assignments.append(AssignmentToSingleColumn(Column(column_name), expression))
+            assignments.append(AssignmentToSingleColumn(AnyTypeColumn(column_name), expression))
         return ProjectQuery(self, columns, assignments)
 
-    def project_rename(self, *args: AssignmentFromColumnToColumn, **kwargs: Column) -> 'ProjectRenameQuery':
+    def project_rename(self, *args: AssignmentFromColumnToColumn, **kwargs: AnyTypeColumn) -> 'ProjectRenameQuery':
         assignments: List[AssignmentFromColumnToColumn] = list(args)
         for column_name, column in kwargs.items():
-            assignments.append(AssignmentFromColumnToColumn(Column(column_name), column))
+            assignments.append(AssignmentFromColumnToColumn(AnyTypeColumn(column_name), column))
         return ProjectRenameQuery(self, assignments)
 
     def project_away(self, *columns: StringType) -> 'ProjectAwayQuery':
@@ -148,9 +149,9 @@ class Query:
                 assignments.append(arg)
         for column_name, expression in kwargs.items():
             if isinstance(expression, BaseExpression):
-                assignments.append(expression.assign_to(Column(column_name)))
+                assignments.append(expression.assign_to(AnyTypeColumn(column_name)))
             else:
-                assignments.append(AnyExpression(to_kql(expression)).assign_to(Column(column_name)))
+                assignments.append(AnyExpression(to_kql(expression)).assign_to(AnyTypeColumn(column_name)))
         return ExtendQuery(self, *assignments)
 
     def summarize(self, *args: Union[AggregationExpression, AssignmentFromAggregationToColumn],
@@ -164,10 +165,10 @@ class Query:
             else:
                 raise ValueError("Invalid assignment: " + str(arg))
         for column_name, agg in kwargs.items():
-            assignments.append(AssignmentFromAggregationToColumn(Column(column_name), agg))
+            assignments.append(AssignmentFromAggregationToColumn(AnyTypeColumn(column_name), agg))
         return SummarizeQuery(self, assignments)
 
-    def mv_expand(self, *columns: Union[Column, ColumnToType], bag_expansion: BagExpansion = None, with_item_index: Column = None,
+    def mv_expand(self, *columns: Union[AnyTypeColumn, ColumnToType], bag_expansion: BagExpansion = None, with_item_index: AnyTypeColumn = None,
                   limit: int = None) -> 'MvExpandQuery':
         if len(columns) == 0:
             raise ValueError("Please specify one or more columns for mv-expand")
@@ -179,15 +180,16 @@ class Query:
     def evaluate(self, plugin_name, *args: ExpressionType, distribution: Distribution = None) -> 'EvaluateQuery':
         return EvaluateQuery(self, plugin_name, *args, distribution=distribution)
 
-    def evaluate_udf(self, udf: FunctionType, extend: bool = True, distribution: Distribution = None, **type_specs: TypeName) -> 'EvaluateQuery':
+    def evaluate_udf(self, udf: FunctionType, extend: bool = True, distribution: Distribution = None, **type_specs: KustoType) -> 'EvaluateQuery':
         return EvaluateQuery(
             self, 'python',
-            BaseExpression(KQL('typeof({})'.format(('*, ' if extend else '') + ', '.join(field_name + ':' + type_name.value for field_name, type_name in type_specs.items())))),
+            BaseExpression(
+                KQL('typeof({})'.format(('*, ' if extend else '') + ', '.join(field_name + ':' + kusto_type.primary_name for field_name, kusto_type in type_specs.items())))),
             stringify_python_func(udf),
             distribution=distribution
         )
 
-    def bag_unpack(self, col: Column, prefix: str = None) -> 'EvaluateQuery':
+    def bag_unpack(self, col: AnyTypeColumn, prefix: str = None) -> 'EvaluateQuery':
         if prefix is None:
             return EvaluateQuery(self, 'bag_unpack', col)
         return EvaluateQuery(self, 'bag_unpack', col, prefix)
@@ -240,10 +242,10 @@ class Query:
 
 
 class ProjectQuery(Query):
-    _columns: List[Column]
+    _columns: List[AnyTypeColumn]
     _assignments: List[AssignmentBase]
 
-    def __init__(self, head: 'Query', columns: List[Column], assignments: List[AssignmentBase]) -> None:
+    def __init__(self, head: 'Query', columns: List[AnyTypeColumn], assignments: List[AssignmentBase]) -> None:
         super().__init__(head)
         self._columns = columns
         self._assignments = assignments
@@ -275,7 +277,7 @@ class ProjectAwayQuery(Query):
 
     def _compile(self) -> KQL:
         def col_or_wildcard_to_string(col_or_wildcard):
-            return col_or_wildcard.kql if isinstance(col_or_wildcard, Column) else col_or_wildcard
+            return col_or_wildcard.kql if isinstance(col_or_wildcard, AnyTypeColumn) else col_or_wildcard
 
         return KQL('project-away {}'.format(', '.join((col_or_wildcard_to_string(c) for c in self._columns))))
 
@@ -410,7 +412,7 @@ class TopQuery(Query):
     _num_rows: int
     _order_spec: OrderQuery.OrderSpec
 
-    def __init__(self, head: Query, num_rows: int, col: Column, order: Order, nulls: Nulls):
+    def __init__(self, head: Query, num_rows: int, col: AnyTypeColumn, order: Order, nulls: Nulls):
         super(TopQuery, self).__init__(head)
         self._num_rows = num_rows
         self._order_spec = OrderQuery.OrderSpec(col, order, nulls)
@@ -427,21 +429,21 @@ class JoinException(Exception):
 class JoinQuery(Query):
     _joined_query: Query
     _kind: JoinKind
-    _on_attributes: Tuple[Tuple[Column, ...], ...]
+    _on_attributes: Tuple[Tuple[AnyTypeColumn, ...], ...]
 
     def __init__(self, head: Query, joined_query: Query, kind: JoinKind,
-                 on_attributes: Tuple[Tuple[Column, ...], ...] = tuple()):
+                 on_attributes: Tuple[Tuple[AnyTypeColumn, ...], ...] = tuple()):
         super(JoinQuery, self).__init__(head)
         self._joined_query = joined_query
         self._kind = kind
         self._on_attributes = on_attributes
 
-    def on(self, col1: Column, col2: Column = None) -> 'JoinQuery':
+    def on(self, col1: AnyTypeColumn, col2: AnyTypeColumn = None) -> 'JoinQuery':
         self._on_attributes = self._on_attributes + (((col1,),) if col2 is None else ((col1, col2),))
         return self
 
     @staticmethod
-    def _compile_on_attribute(attribute: Tuple[Column]):
+    def _compile_on_attribute(attribute: Tuple[AnyTypeColumn]):
         assert len(attribute) in (1, 2)
         if len(attribute) == 1:
             return attribute[0].kql
@@ -462,7 +464,7 @@ class JoinQuery(Query):
 
 class SummarizeQuery(Query):
     _assignments: List[AssignmentFromAggregationToColumn]
-    _by_columns: List[Union[Column, BaseExpression]]
+    _by_columns: List[Union[AnyTypeColumn, BaseExpression]]
     _by_assignments: List[AssignmentToSingleColumn]
 
     def __init__(self, head: Query,
@@ -472,17 +474,17 @@ class SummarizeQuery(Query):
         self._by_columns = []
         self._by_assignments = []
 
-    def by(self, *args: Union[AssignmentToSingleColumn, Column, BaseExpression],
+    def by(self, *args: Union[AssignmentToSingleColumn, AnyTypeColumn, BaseExpression],
            **kwargs: BaseExpression):
         for arg in args:
-            if isinstance(arg, Column) or isinstance(arg, BaseExpression):
+            if isinstance(arg, AnyTypeColumn) or isinstance(arg, BaseExpression):
                 self._by_columns.append(arg)
             elif isinstance(arg, AssignmentToSingleColumn):
                 self._by_assignments.append(arg)
             else:
                 raise ValueError("Invalid assignment: " + arg.to_kql())
         for column_name, group_exp in kwargs.items():
-            self._by_assignments.append(AssignmentToSingleColumn(Column(column_name), group_exp))
+            self._by_assignments.append(AssignmentToSingleColumn(AnyTypeColumn(column_name), group_exp))
         return self
 
     def _compile(self) -> KQL:
@@ -496,12 +498,12 @@ class SummarizeQuery(Query):
 
 
 class MvExpandQuery(Query):
-    _columns: Tuple[Union[Column, ColumnToType]]
+    _columns: Tuple[Union[AnyTypeColumn, ColumnToType]]
     _bag_expansion: BagExpansion
-    _with_item_index: Column
+    _with_item_index: AnyTypeColumn
     _limit: int
 
-    def __init__(self, head: Query, columns: Tuple[Union[Column, ColumnToType]], bag_expansion: BagExpansion, with_item_index: Column, limit: int):
+    def __init__(self, head: Query, columns: Tuple[Union[AnyTypeColumn, ColumnToType]], bag_expansion: BagExpansion, with_item_index: AnyTypeColumn, limit: int):
         super(MvExpandQuery, self).__init__(head)
         self._columns = columns
         self._bag_expansion = bag_expansion
diff --git a/pykusto/type_utils.py b/pykusto/type_utils.py
index 7a5daa2..afef29b 100644
--- a/pykusto/type_utils.py
+++ b/pykusto/type_utils.py
@@ -1,26 +1,69 @@
 from datetime import datetime, timedelta
 from enum import Enum
-from numbers import Number
 from typing import Union, Mapping, Type, Dict, Callable, Tuple, List, Any, Set
 
-KustoTypes = Union[str, Number, bool, datetime, Mapping, List, Tuple, timedelta]
 # TODO: Unhandled data types: guid, decimal
-
-
-def get_base_types(obj: Any) -> Set[Type[KustoTypes]]:
+PythonTypes = Union[str, int, float, bool, datetime, Mapping, List, Tuple, timedelta]
+
+
+class KustoType(Enum):
+    BOOL = ('bool', 'I8', 'System.SByte', bool)
+    DATETIME = ('datetime', 'DateTime', 'System.DateTime', datetime)
+    # noinspection PyTypeChecker
+    DECIMAL = ('decimal', 'Decimal', 'System.Data.SqlTypes.SqlDecimal', None)  # TODO
+    ARRAY = ('dynamic', 'Dynamic', 'System.Object', List, Tuple)
+    MAPPING = ('dynamic', 'Dynamic', 'System.Object', Mapping)
+    # noinspection PyTypeChecker
+    GUID = ('guid', 'UniqueId', 'System.Guid', None)  # TODO
+    INT = ('int', 'I32', 'System.Int32', int)
+    LONG = ('long', 'I64', 'System.Int64', int)
+    REAL = ('real', 'R64', 'System.Double', float)
+    STRING = ('string', 'StringBuffer', 'System.String', str)
+    TIMESPAN = ('timespan', 'TimeSpan', 'System.TimeSpan', timedelta)
+    NULL = ('null', 'null', 'null', type(None))
+
+    primary_name: str
+    internal_name: str
+    dot_net_name: str
+    python_types: Tuple[Type[PythonTypes]]
+
+    def __init__(self, primary_name: str, internal_name: str, dot_net_name: str, *python_types: Type[PythonTypes]) -> None:
+        self.primary_name = primary_name
+        self.internal_name = internal_name
+        self.dot_net_name = dot_net_name
+        self.python_types = python_types
+
+    def is_type_of(self, obj) -> bool:
+        for python_type in self.python_types:
+            if python_type is not None and isinstance(obj, python_type):
+                return True
+        return False
+
+    def is_superclass_of(self, t: Type) -> bool:
+        for python_type in self.python_types:
+            if python_type is not None and issubclass(t, python_type):
+                return True
+        return False
+
+
+INTERNAL_NAME_TO_TYPE: Dict[str, KustoType] = {t.internal_name: t for t in KustoType}
+DOT_NAME_TO_TYPE: Dict[str, KustoType] = {t.dot_net_name: t for t in KustoType}
+
+
+def get_base_types(obj: Any) -> Set[KustoType]:
     """
-    For a given object, return the associated basic type, which is a member of `KustoTypes`
+    For a given object, return the associated basic type, which is a member of :class:`KustoType`
 
     :param obj: The given object for which the type is resolved
-    :return: A type which is a member of `KustoTypes`
+    :return: A type which is a member of `KustoType`
     """
-    obj_type = type(obj)
-    for kusto_type in KustoTypes.__args__:
-        if isinstance(obj, kusto_type):
+    for kusto_type in KustoType:
+        if kusto_type.is_type_of(obj):
             # The object is already a member of Kusto types
             return {kusto_type}
     # The object is one of the expression types decorated with a TypeRegistrar, therefore the original types are
     # recorded the field _base_types
+    obj_type = type(obj)
     base_types = getattr(obj_type, '_base_types', None)
     if base_types is None:
         raise TypeError("get_base_types called for unsupported type: {}".format(obj_type.__name__))
@@ -29,11 +72,12 @@ def get_base_types(obj: Any) -> Set[Type[KustoTypes]]:
 
 class TypeRegistrar:
     """
-    A factory for annotations that are used to create a mapping between Kusto types and python types and functions.
+    A factory for annotations that are used to create a mapping between Kusto types and python types / functions.
     Each annotation must be called with a Kusto type as a parameter. The `for_obj` and `for_type` methods
     can then be used to retrieve the python type or function corresponding to a given Kusto type.
     """
-    registry: Dict[Type[KustoTypes], Callable]
+    name: str
+    registry: Dict[KustoType, Callable]
 
     def __init__(self, name: str) -> None:
         """
@@ -45,18 +89,18 @@ class TypeRegistrar:
     def __repr__(self) -> str:
         return self.name
 
-    def __call__(self, *types: Type[KustoTypes]) -> Callable:
+    def __call__(self, *types: KustoType) -> Callable:
         def inner(wrapped):
             for t in types:
                 previous = self.registry.setdefault(t, wrapped)
                 if previous is not wrapped:
-                    raise TypeError("{}: type already registered: {}".format(self, t.__name__))
+                    raise TypeError("{}: type already registered: {}".format(self, t.primary_name))
             wrapped._base_types = set(types)
             return wrapped
 
         return inner
 
-    def for_obj(self, obj: KustoTypes) -> Callable:
+    def for_obj(self, obj: PythonTypes) -> Callable:
         """
         Given an object of Kusto type, retrieve the python type or function associated with the object's type, and call
         it with the given object as a parameter
@@ -65,11 +109,11 @@ class TypeRegistrar:
         :return: Associated python object
         """
         for registered_type, registered_callable in self.registry.items():
-            if isinstance(obj, registered_type):
+            if registered_type.is_type_of(obj):
                 return registered_callable(obj)
         raise ValueError("{}: no registered callable for object {} of type {}".format(self, obj, type(obj).__name__))
 
-    def for_type(self, t: Type[KustoTypes]) -> Callable:
+    def for_type(self, t: Type[PythonTypes]) -> Callable:
         """
         Given a Kusto type, retrieve the associated python type or function
 
@@ -77,24 +121,12 @@ class TypeRegistrar:
         :return: Associated python object
         """
         for registered_type, registered_callable in self.registry.items():
-            if issubclass(t, registered_type):
+            if registered_type.is_superclass_of(t):
                 return registered_callable
         raise ValueError("{}: no registered callable for type {}".format(self, t.__name__))
 
 
 kql_converter = TypeRegistrar("KQL Converter")
+typed_column = TypeRegistrar("Column")
 plain_expression = TypeRegistrar("Plain expression")
 aggregation_expression = TypeRegistrar("Aggregation expression")
-
-
-class TypeName(Enum):
-    BOOL = 'bool'
-    DATETIME = 'datetime'
-    DECIMAL = 'decimal'
-    DYNAMIC = 'dynamic'
-    GUID = 'guid'
-    INT = 'int'
-    LONG = 'long'
-    REAL = 'real'
-    STRING = 'string'
-    TIMESPAN = 'timespan'
diff --git a/setup.py b/setup.py
index 9fa5b79..f075904 100644
--- a/setup.py
+++ b/setup.py
@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
 
 setup(
     name='pykusto',
-    version='0.0.10',
+    version='0.0.11',
     packages=find_packages(exclude=['test']),
     url='https://github.com/Azure/pykusto',
     license='MIT License',


Test patch:
diff --git a/test/test_functions.py b/test/test_functions.py
index b88182a..c23866e 100644
--- a/test/test_functions.py
+++ b/test/test_functions.py
@@ -738,7 +738,7 @@ class TestFunction(TestBase):
 
     def test_iff_different_types(self):
         self.assertRaises(
-            TypeError("The second and third arguments must be of the same type, but they are: timedelta and str"),
+            TypeError("The second and third arguments must be of the same type, but they are: timespan and string"),
             lambda: Query().project(foo=f.iff(col.foo > f.ago(datetime.timedelta(2)), datetime.timedelta(3), "hello"))
                 .render()
         )
diff --git a/test/test_query.py b/test/test_query.py
index c28e76c..4ec926d 100644
--- a/test/test_query.py
+++ b/test/test_query.py
@@ -2,7 +2,7 @@ from pykusto import functions as f
 from pykusto.client import PyKustoClient
 from pykusto.expressions import column_generator as col
 from pykusto.query import Query, Order, Nulls, JoinKind, JoinException, BagExpansion, Distribution
-from pykusto.type_utils import TypeName
+from pykusto.type_utils import KustoType
 from test.test_base import TestBase
 from test.test_table import MockKustoClient
 from test.udf import func, STRINGIFIED
@@ -163,6 +163,12 @@ class TestQuery(TestBase):
             Query().extend(col.foo.assign_to(col.foo1, col.foo2), shoo=col.bar * 4).render(),
         )
 
+    def test_extend_assign_non_array_to_multiple_columns(self):
+        self.assertRaises(
+            ValueError("Only arrays can be assigned to multiple columns"),
+            lambda: Query().extend(f.strcat("hello", "world").assign_to(col.foo1, col.foo2)).render(),
+        )
+
     def test_extend_generate_column_name(self):
         self.assertEqual(
             " | extend v1 + v2, foo = bar * 4",
@@ -203,7 +209,7 @@ class TestQuery(TestBase):
     def test_mv_expand_to_type(self):
         self.assertEqual(
             " | mv-expand a to typeof(string), b to typeof(int), c",
-            Query().mv_expand(f.to_type(col.a, TypeName.STRING), f.to_type(col.b, TypeName.INT), col.c).render(),
+            Query().mv_expand(f.to_type(col.a, KustoType.STRING), f.to_type(col.b, KustoType.INT), col.c).render(),
         )
 
     def test_mv_expand_args(self):
@@ -312,13 +318,13 @@ class TestQuery(TestBase):
     def test_udf(self):
         self.assertEqual(
             " | evaluate python(typeof(*, StateZone:string), {})".format(STRINGIFIED),
-            Query().evaluate_udf(func, StateZone=TypeName.STRING).render(),
+            Query().evaluate_udf(func, StateZone=KustoType.STRING).render(),
         )
 
     def test_udf_no_extend(self):
         self.assertEqual(
             " | evaluate python(typeof(StateZone:string), {})".format(STRINGIFIED),
-            Query().evaluate_udf(func, extend=False, StateZone=TypeName.STRING).render(),
+            Query().evaluate_udf(func, extend=False, StateZone=KustoType.STRING).render(),
         )
 
     def test_bag_unpack(self):
diff --git a/test/test_table.py b/test/test_table.py
index f817699..9ce0205 100644
--- a/test/test_table.py
+++ b/test/test_table.py
@@ -1,24 +1,73 @@
-from typing import List, Tuple
+from concurrent.futures import Future
+from typing import List, Tuple, Callable, Any
 from unittest.mock import patch
 from urllib.parse import urljoin
 
 from azure.kusto.data.request import KustoClient, ClientRequestProperties
 
-from pykusto.expressions import column_generator as col
 from pykusto.client import PyKustoClient
+from pykusto.expressions import column_generator as col, StringColumn, NumberColumn, AnyTypeColumn, BooleanColumn
 from pykusto.query import Query
+from pykusto.type_utils import KustoType
 from test.test_base import TestBase
 
 
+def mock_response(rows: Tuple[Any, ...]):
+    return type(
+        'KustoResponseDataSet',
+        (object,),
+        {'primary_results': (type(
+            'KustoResultTable',
+            (object,),
+            {'rows': rows}
+        ),)}
+    )
+
+
+def mock_columns_response(columns: List[Tuple[str, KustoType]] = tuple()) -> Callable:
+    return lambda: mock_response(tuple((c_name, c_type.internal_name) for c_name, c_type in columns))
+
+
+def mock_tables_response(tables: List[Tuple[str, List[Tuple[str, KustoType]]]] = tuple()) -> Callable:
+    return lambda: mock_response(tuple((t_name, c_name, c_type.dot_net_name) for t_name, columns in tables for c_name, c_type in columns))
+
+
+def mock_databases_response(databases: List[Tuple[str, List[Tuple[str, List[Tuple[str, KustoType]]]]]] = tuple()) -> Callable:
+    return lambda: mock_response(tuple(
+        (d_name, t_name, c_name, c_type.dot_net_name)
+        for d_name, tables in databases
+        for t_name, columns in tables
+        for c_name, c_type in columns
+    ))
+
+
 # noinspection PyMissingConstructor
 class MockKustoClient(KustoClient):
     executions: List[Tuple[str, str, ClientRequestProperties]]
-
-    def __init__(self, cluster="https://test_cluster.kusto.windows.net"):
+    columns_response: Callable
+    tables_response: Callable
+    databases_response: Callable
+
+    def __init__(
+            self,
+            cluster="https://test_cluster.kusto.windows.net",
+            columns_response: Callable = mock_columns_response([]),
+            tables_response: Callable = mock_tables_response([]),
+            databases_response: Callable = mock_databases_response([]),
+    ):
         self.executions = []
         self._query_endpoint = urljoin(cluster, "/v2/rest/query")
+        self.columns_response = columns_response
+        self.tables_response = tables_response
+        self.databases_response = databases_response
 
     def execute(self, database: str, rendered_query: str, properties: ClientRequestProperties = None):
+        if rendered_query == '.show database schema | project TableName, ColumnName, ColumnType | limit 10000':
+            return self.tables_response()
+        if rendered_query.startswith('.show table '):
+            return self.columns_response()
+        if rendered_query.startswith('.show databases schema '):
+            return self.databases_response()
         self.executions.append((database, rendered_query, properties))
 
 
@@ -128,3 +177,76 @@ class TestTable(TestBase):
               None)],
             mock_kusto_client_1.executions,
         )
+
+    def test_column_retrieve(self):
+        mock_kusto_client = MockKustoClient(columns_response=mock_columns_response([('foo', KustoType.STRING), ('bar', KustoType.INT)]))
+        table = PyKustoClient(mock_kusto_client)['test_db']['test_table']
+        table.refresh()
+        table.wait_for_items()  # Avoid race condition
+        self.assertEqual(StringColumn, type(table.foo))
+        self.assertEqual(NumberColumn, type(table.bar))
+        self.assertEqual(AnyTypeColumn, type(table['baz']))
+
+    def test_column_retrieve_brackets(self):
+        mock_kusto_client = MockKustoClient(columns_response=mock_columns_response([('foo', KustoType.STRING), ('bar', KustoType.INT)]))
+        table = PyKustoClient(mock_kusto_client)['test_db']['test_table']
+        table.refresh()
+        table.wait_for_items()  # Avoid race condition
+        self.assertEqual(StringColumn, type(table['foo']))
+        self.assertEqual(NumberColumn, type(table['bar']))
+        self.assertEqual(AnyTypeColumn, type(table['baz']))
+
+    def test_column_retrieve_slow(self):
+        mock_response_future = Future()
+        try:
+            mock_kusto_client = MockKustoClient(columns_response=lambda: mock_response_future.result())
+            table = PyKustoClient(mock_kusto_client)['test_db']['test_table']
+            self.assertEqual(AnyTypeColumn, type(table['foo']))
+            self.assertEqual(AnyTypeColumn, type(table['bar']))
+            self.assertEqual(AnyTypeColumn, type(table['baz']))
+        finally:
+            mock_response_future.set_result(mock_columns_response([])())
+
+    def test_table_retrieve(self):
+        mock_kusto_client = MockKustoClient(tables_response=mock_tables_response([('test_table', [('foo', KustoType.STRING), ('bar', KustoType.INT)])]))
+        db = PyKustoClient(mock_kusto_client)['test_db']
+        db.refresh()
+        db.wait_for_items()  # Avoid race condition
+        table = db.test_table
+        self.assertEqual(StringColumn, type(table.foo))
+        self.assertEqual(NumberColumn, type(table.bar))
+        self.assertEqual(AnyTypeColumn, type(table['baz']))
+        self.assertEqual(AnyTypeColumn, type(db['other_table']['foo']))
+
+    def test_table_retrieve_error(self):
+        mock_kusto_client = MockKustoClient(tables_response=mock_tables_response([('test_table', [('foo', KustoType.STRING), ('bar', KustoType.INT)])]))
+        db = PyKustoClient(mock_kusto_client)['test_db']
+        db.refresh()
+        db.wait_for_items()  # Avoid race condition
+        self.assertRaises(
+            AttributeError("PyKustoClient(test_cluster.kusto.windows.net).Database(test_db) has no attribute 'test_table_1'"),
+            lambda: db.test_table_1
+        )
+
+    def test_two_tables_retrieve(self):
+        mock_kusto_client = MockKustoClient(tables_response=mock_tables_response([
+            ('test_table_1', [('foo', KustoType.STRING), ('bar', KustoType.INT)]),
+            ('test_table_2', [('baz', KustoType.BOOL)])
+        ]))
+        db = PyKustoClient(mock_kusto_client)['test_db']
+        db.refresh()
+        db.wait_for_items()  # Avoid race condition
+        self.assertEqual(StringColumn, type(db.test_table_1.foo))
+        self.assertEqual(NumberColumn, type(db.test_table_1.bar))
+        self.assertEqual(BooleanColumn, type(db.test_table_2['baz']))
+        self.assertEqual(AnyTypeColumn, type(db['other_table']['foo']))
+
+    def test_database_retrieve(self):
+        mock_kusto_client = MockKustoClient(databases_response=mock_databases_response([('test_db', [('test_table', [('foo', KustoType.STRING), ('bar', KustoType.INT)])])]))
+        client = PyKustoClient(mock_kusto_client)
+        client.wait_for_items()
+        table = client.test_db.test_table
+        self.assertEqual(StringColumn, type(table.foo))
+        self.assertEqual(NumberColumn, type(table.bar))
+        self.assertEqual(AnyTypeColumn, type(table['baz']))
+        self.assertEqual(AnyTypeColumn, type(client.test_db['other_table']['foo']))
diff --git a/test/test_utils.py b/test/test_utils.py
index 1cc326d..30ae78a 100644
--- a/test/test_utils.py
+++ b/test/test_utils.py
@@ -1,5 +1,5 @@
 from pykusto.expressions import to_kql
-from pykusto.type_utils import TypeRegistrar
+from pykusto.type_utils import TypeRegistrar, KustoType
 from test.test_base import TestBase
 
 
@@ -20,7 +20,7 @@ class TestUtils(TestBase):
     def test_type_registrar_for_type(self):
         test_annotation = TypeRegistrar("Test annotation")
 
-        @test_annotation(str)
+        @test_annotation(KustoType.STRING)
         def str_annotated(s: str) -> str:
             return "response to " + s
 
@@ -32,7 +32,7 @@ class TestUtils(TestBase):
     def test_type_registrar_for_obj(self):
         test_annotation = TypeRegistrar("Test annotation")
 
-        @test_annotation(str)
+        @test_annotation(KustoType.STRING)
         def str_annotated(s: str) -> str:
             return "response to " + s
 
@@ -44,7 +44,7 @@ class TestUtils(TestBase):
     def test_type_registrar_for_type_not_found(self):
         test_annotation = TypeRegistrar("Test annotation")
 
-        @test_annotation(str)
+        @test_annotation(KustoType.STRING)
         def str_annotated(s: str) -> str:
             return "response to " + s
 
@@ -56,7 +56,7 @@ class TestUtils(TestBase):
     def test_type_registrar_for_obj_not_found(self):
         test_annotation = TypeRegistrar("Test annotation")
 
-        @test_annotation(str)
+        @test_annotation(KustoType.STRING)
         def str_annotated(s: str) -> str:
             return "response to " + s
 
@@ -68,7 +68,7 @@ class TestUtils(TestBase):
     def test_type_registrar_collision(self):
         test_annotation = TypeRegistrar("Test annotation")
 
-        @test_annotation(str)
+        @test_annotation(KustoType.STRING)
         def str_annotated_1(s: str) -> str:
             return "response to " + s
 
@@ -76,6 +76,6 @@ class TestUtils(TestBase):
             return "response to " + s
 
         self.assertRaises(
-            TypeError("Test annotation: type already registered: str"),
-            lambda: test_annotation(str)(str_annotated_2)
+            TypeError("Test annotation: type already registered: string"),
+            lambda: test_annotation(KustoType.STRING)(str_annotated_2)
         )