diff --git a/TODO b/TODO
index 67ea435..4bbc177 100644
--- a/TODO
+++ b/TODO
@@ -21,6 +21,8 @@ Types:
 
   -- need List<T>, Map<T1, T2> types
 
+  -- BINARY type should be able to specify encoding when converting to STRING.
+
 Aggregation:
 
   -- unit test, unit test, unit test, baby.
diff --git a/src/main/antlr3/com/odiago/flumebase/parser/SqlGrammar.g b/src/main/antlr3/com/odiago/flumebase/parser/SqlGrammar.g
index 224943a..2b57df1 100644
--- a/src/main/antlr3/com/odiago/flumebase/parser/SqlGrammar.g
+++ b/src/main/antlr3/com/odiago/flumebase/parser/SqlGrammar.g
@@ -247,6 +247,7 @@ primitive_field_type returns [String val]:
   | FLOAT { $val = "FLOAT"; }
   | DOUBLE { $val = "DOUBLE"; }
   | STRING_KW { $val = "STRING"; }
+  | BINARY { $val = "BINARY"; }
   | TIMESTAMP { $val = "TIMESTAMP"; };
 
 // Defines a PRECISE(n) type.
diff --git a/src/main/antlr3/com/odiago/flumebase/parser/SqlLexer.g b/src/main/antlr3/com/odiago/flumebase/parser/SqlLexer.g
index aa64dff..0606225 100644
--- a/src/main/antlr3/com/odiago/flumebase/parser/SqlLexer.g
+++ b/src/main/antlr3/com/odiago/flumebase/parser/SqlLexer.g
@@ -30,6 +30,7 @@ L_AND : A N D ;
 AS : A S ;
 BETWEEN : B E T W E E N ;
 BIGINT : B I G I N T ;
+BINARY : B I N A R Y ;
 BOOLEAN : B O O L E A N ;
 BY : B Y ;
 CREATE : C R E A T E ;
diff --git a/src/main/java/com/odiago/flumebase/io/CachingTextEventParser.java b/src/main/java/com/odiago/flumebase/io/CachingTextEventParser.java
index d121af7..eaf32e5 100644
--- a/src/main/java/com/odiago/flumebase/io/CachingTextEventParser.java
+++ b/src/main/java/com/odiago/flumebase/io/CachingTextEventParser.java
@@ -121,6 +121,9 @@ protected Object parseAndCache(CharBuffer chars, int colIdx, Type expectedType)
     // TODO(aaron): Test how this handles a field that is an empty string.
     Object out = null;
     switch (primitiveTypeName) {
+    case BINARY:
+      out = /* TODO: BASE64-decode the chars */ bla bla bla; 
+      break;
     case BOOLEAN:
       out = CharBufferUtils.parseBool(chars);
       break;
diff --git a/src/main/java/com/odiago/flumebase/lang/Type.java b/src/main/java/com/odiago/flumebase/lang/Type.java
index c6f4422..3f93b1b 100644
--- a/src/main/java/com/odiago/flumebase/lang/Type.java
+++ b/src/main/java/com/odiago/flumebase/lang/Type.java
@@ -50,6 +50,7 @@ public enum TypeName implements Comparable<TypeName> {
     PRECISE(5), // A numeric type which allows the user to specify an arbitrary
                 // degree of precision.
     STRING(8),
+    BINARY(6),
     TIMESTAMP(6),
     TIMESPAN(7),
     ANY(0), // 'null' constant can be cast to any type. Only valid inside NULLABLE.
@@ -140,6 +141,7 @@ protected Type(TypeName name) {
     PRIMITIVE_TYPES.put(TypeName.FLOAT, new Type(TypeName.FLOAT));
     PRIMITIVE_TYPES.put(TypeName.DOUBLE, new Type(TypeName.DOUBLE));
     PRIMITIVE_TYPES.put(TypeName.STRING, new Type(TypeName.STRING));
+    PRIMITIVE_TYPES.put(TypeName.BINARY, new Type(TypeName.BINARY));
     PRIMITIVE_TYPES.put(TypeName.TIMESTAMP, new Type(TypeName.TIMESTAMP));
     PRIMITIVE_TYPES.put(TypeName.TIMESPAN, new Type(TypeName.TIMESPAN));
     PRIMITIVE_TYPES.put(TypeName.TYPECLASS_NUMERIC, new Type(TypeName.TYPECLASS_NUMERIC));
@@ -156,6 +158,7 @@ protected Type(TypeName name) {
     NULLABLE_TYPES.put(TypeName.FLOAT, new NullableType(TypeName.FLOAT));
     NULLABLE_TYPES.put(TypeName.DOUBLE, new NullableType(TypeName.DOUBLE));
     NULLABLE_TYPES.put(TypeName.STRING, new NullableType(TypeName.STRING));
+    NULLABLE_TYPES.put(TypeName.BINARY, new NullableType(TypeName.BINARY));
     NULLABLE_TYPES.put(TypeName.TIMESTAMP, new NullableType(TypeName.TIMESTAMP));
     NULLABLE_TYPES.put(TypeName.TIMESPAN, new NullableType(TypeName.TIMESPAN));
     NULLABLE_TYPES.put(TypeName.ANY, new NullableType(TypeName.ANY));
@@ -253,6 +256,7 @@ public boolean isScalar() {
     case DOUBLE:
     case PRECISE:
     case STRING:
+    case BINARY:
     case TIMESTAMP:
     case TIMESPAN:
     case WINDOW:
@@ -268,6 +272,7 @@ public boolean isScalar() {
   public boolean isComparable() {
     return isNumeric() || this.equals(Type.getPrimitive(TypeName.BOOLEAN))
         || this.equals(Type.getPrimitive(TypeName.STRING))
+        || this.equals(Type.getPrimitive(TypeName.BINARY))
         || this.equals(Type.getPrimitive(TypeName.TYPECLASS_COMPARABLE));
   }
 
@@ -323,6 +328,8 @@ public Schema getAvroSchema() {
    *                                                with the exception of "WINDOW".)
    *
    *</tt></pre></li>
+   *   <li>The lattice ANY -&gt; BINARY -&gt; STRING -&gt; ...  also holds.
+   *     Bytes are BASE64-encoded when converted to strings.</li>
    *   <li>An additional rule governs meets over PRECISE(k) types:
    *     meet(PRECISE(n), PRECISE(m)) = PRECISE(MAX(n, m))</li>
    *   <li>The following lattice describes TYPECLASS_COMPARABLE and TYPECLASS_ANY:<pre><tt>
@@ -563,6 +570,8 @@ protected Schema getAvroSchema(TypeName typeName) {
       return Schema.create(Schema.Type.DOUBLE);
     case STRING:
       return Schema.create(Schema.Type.STRING);
+    case BINARY:
+      return Schema.create(Schema.Type.BYTES);
     case TIMESPAN:
       return TimeSpan.SCHEMA$;
     case TIMESTAMP:
diff --git a/src/test/java/com/odiago/flumebase/lang/TestType.java b/src/test/java/com/odiago/flumebase/lang/TestType.java
index 4773556..a38326e 100644
--- a/src/test/java/com/odiago/flumebase/lang/TestType.java
+++ b/src/test/java/com/odiago/flumebase/lang/TestType.java
@@ -75,6 +75,19 @@ public void testPromotion() {
         Type.getNullable(Type.TypeName.FLOAT)));
     assertTrue(Type.getPrimitive(Type.TypeName.INT).promotesTo(
         Type.getNullable(Type.TypeName.STRING)));
+
+    // BINARY type
+    assertTrue(Type.getPrimitive(Type.TypeName.BINARY).promotesTo(
+        Type.getNullable(Type.TypeName.BINARY)));
+    assertTrue(Type.getPrimitive(Type.TypeName.BINARY).promotesTo(
+        Type.getNullable(Type.TypeName.STRING)));
+
+    assertFalse(Type.getPrimitive(Type.TypeName.BINARY).promotesTo(
+        Type.getPrimitive(Type.TypeName.INT)));
+    assertFalse(Type.getPrimitive(Type.TypeName.INT).promotesTo(
+        Type.getPrimitive(Type.TypeName.BINARY)));
+    assertFalse(Type.getPrimitive(Type.TypeName.BINARY).promotesTo(
+        Type.getNullable(Type.TypeName.INT)));
   
     // STRING to NULLABLE(STRING).
     assertTrue(Type.getPrimitive(Type.TypeName.STRING).promotesTo(