Cascading type coercion difficulties

test output

/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/bin/java -ea -Didea.launcher.port=7532 "-Didea.launcher.bin.path=/Applications/IntelliJ IDEA 13 CE.app/bin" -Dfile.encoding=UTF-8 -classpath "/Applications/IntelliJ IDEA 13 CE.app/lib/idea_rt.jar:/Applications/IntelliJ IDEA 13 CE.app/plugins/junit/lib/junit-rt.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/lib/ant-javafx.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/lib/dt.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/lib/javafx-doclet.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/lib/javafx-mx.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/lib/jconsole.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/lib/sa-jdi.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/lib/tools.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/charsets.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/deploy.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/htmlconverter.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/javaws.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/jce.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/jfr.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/jfxrt.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/jsse.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/management-agent.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/plugin.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/resources.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/rt.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/ext/dnsns.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/ext/localedata.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/ext/sunec.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/ext/sunjce_provider.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/ext/sunpkcs11.jar:/Library/Java/JavaVirtualMachines/jdk1.7.0_51.jdk/Contents/Home/jre/lib/ext/zipfs.jar:/Users/luis.casillas/Source/analytics-engineering/cascading-utils/build/classes/test:/Users/luis.casillas/Source/analytics-engineering/cascading-utils/build/classes/main:/Users/luis.casillas/Source/analytics-engineering/testutils/build/classes/test:/Users/luis.casillas/Source/analytics-engineering/testutils/build/classes/main:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/com.google.guava/guava/16.0.1/5fa98cd1a63c99a44dd8d3b77e4762b066a5d0c5/guava-16.0.1.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/junit/junit/4.11/4e031bb61df09069aeb2bffb4019e7a5034a4ee0/junit-4.11.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.junit.contrib/junit-theories/4.11/2a8ff166840746582ef6cc8fa6d949ebbc7dc4f1/junit-theories-4.11.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/com.pholser/junit-quickcheck-core/0.3/d9f23975cc05dbd6b1b342669c3d817704dcf3a4/junit-quickcheck-core-0.3.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/com.pholser/junit-quickcheck-generators/0.3/2305a86f1f1c4f829cfa7e5c304d6de465b13655/junit-quickcheck-generators-0.3.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/com.pholser/junit-quickcheck-guava/0.3/36d66b282a4a8b85d90fd06e56dedb286a692940/junit-quickcheck-guava-0.3.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.hamcrest/hamcrest-all/1.3/63a21ebc981131004ad02e0434e799fd7f3a8d5a/hamcrest-all-1.3.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.hamcrest/hamcrest-core/1.3/42a25dc3219429f0e5d060061f71acb49bf010a0/hamcrest-core-1.3.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.javaruntype/javaruntype/1.2/d7595d6aa87b56ad207961157d15b2c59f0afb53/javaruntype-1.2.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/javassist/javassist/3.11.0.GA/2c00105734a57e9ee4f27e4b17cd43200e5f0ff8/javassist-3.11.0.GA.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/ognl/ognl/3.0.5/e35387ec9e4ca6908dcf184cd4766dd1fa8838d8/ognl-3.0.5.jar:/Users/luis.casillas/Source/analytics-engineering/utilities/build/classes/main:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.antlr/antlr-runtime/3.1.2/c4ca32c2be1b22a5553dd3171f51f9b2b04030b/antlr-runtime-3.1.2.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/com.google.code.findbugs/jsr305/2.0.3/5871fb60dc68d67da54a663c3fd636a10a532948/jsr305-2.0.3.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.hsqldb/hsqldb/2.3.2/970fd7b8f635e2c19305160459649569655b843c/hsqldb-2.3.2.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.hsqldb/sqltool/2.3.2/968e67f9a00a8c7db1b0221961aa78198d217758/sqltool-2.3.2.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.slf4j/slf4j-api/1.7.6/562424e36df3d2327e8e9301a76027fca17d54ea/slf4j-api-1.7.6.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/com.mchange/c3p0/0.9.5-pre8/445c1ade96969d822047e517bd2e2db83679c309/c3p0-0.9.5-pre8.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/com.mchange/mchange-commons-java/0.2.7/be3595e4436419f6d0d69f39b2764c750ce688d6/mchange-commons-java-0.2.7.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.slf4j/slf4j-simple/1.7.6/12aaccff7811a8c619bad5954b96823589d2476e/slf4j-simple-1.7.6.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/cascading/lingual-core/1.1.1/ef4380fb91fa30d28dac866f8117f64040cd4cb5/lingual-core-1.1.1.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/riffle/riffle/0.1-dev/9c78a2a023d7f3f1db5e98680df3d8c074abf004/riffle-0.1-dev.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/thirdparty/jgrapht-jdk1.6/0.8.1/5bf27c4007ee8e28bd63469799a05f587ae1d82c/jgrapht-jdk1.6-0.8.1.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/com.fasterxml.jackson.core/jackson-core/2.2.2/d20be6a5ddd6f8cfd36ebf6dea329873a1c41f1b/jackson-core-2.2.2.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/com.fasterxml.jackson.core/jackson-annotations/2.2.2/285cb9c666f0f0f3dd8a1be04e1f457eb7b15113/jackson-annotations-2.2.2.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/com.fasterxml.jackson.core/jackson-databind/2.2.2/3c8f6018eaa72d43b261181e801e6f8676c16ef6/jackson-databind-2.2.2.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/com.fasterxml.jackson.datatype/jackson-datatype-guava/2.2.2/4ae687c2b92cf41a2ccd52a5fc906f7ba29214d6/jackson-datatype-guava-2.2.2.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/commons-io/commons-io/2.4/b1b6ea3b7e4aa4f492509a4952029cd8e48019ad/commons-io-2.4.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/xerces/xercesImpl/2.5.0/c0468bac6d11a07ffc69506003cfedc0ce54e172/xercesImpl-2.5.0.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.apache.ant/ant-launcher/1.7.1/a9cbbcefbbb5e7f97596045268243a8c1c7aafca/ant-launcher-1.7.1.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.apache.ant/ant/1.7.1/1d33711018e7649a8427fff62a87f94f4e7d310f/ant-1.7.1.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/eigenbase/eigenbase-xom/1.3.4/9fbc84f269d04fc0919d6c1e6cdbf0fd4587ad89/eigenbase-xom-1.3.4.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/eigenbase/eigenbase-properties/1.1.4/945753886102390cc7e7a3f730379265ec1c68e0/eigenbase-properties-1.1.4.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/eigenbase/eigenbase-resgen/1.3.6/faa995a6d2ff4cbdbe09e92ddbb0d07e5634bdac/eigenbase-resgen-1.3.6.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/commons-pool/commons-pool/1.5.4/75b6e20c596ed2945a259cea26d7fadd298398e6/commons-pool-1.5.4.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/commons-dbcp/commons-dbcp/1.4/30be73c965cc990b153a100aaaaafcf239f82d39/commons-dbcp-1.4.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/net.hydromatic/linq4j/0.1.9/63b35bf926e71d8bd6ae74ab9ecbf8ca7bd4962f/linq4j-0.1.9.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/net.hydromatic/optiq-core/0.4.12.4/c88b8a87cf547894b562a425ab749effc65891ce/optiq-core-0.4.12.4.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.slf4j/slf4j-api/1.6.1/6f3b8a24bf970f17289b234284c94f43eb42f0e4/slf4j-api-1.6.1.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/cascading/cascading-bind/1.0.0-wip-106/1b2a67ea5c3bea6cacd9e69edc80ad8504b52dd8/cascading-bind-1.0.0-wip-106.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.apache.commons/commons-lang3/3.1/905075e6c80f206bbe6cf1e809d2caa69f420c76/commons-lang3-3.1.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/javassist/javassist/3.12.1.GA/526633327faa61aee448a519e8a4d53ec3057885/javassist-3.12.1.GA.jar:/Users/luis.casillas/Source/analytics-engineering/cascading-utils/src/main/lib/jeospatial.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/com.amazonaws/aws-java-sdk/1.8.2/d5473b4804a394242514c38ca62aceb629713b9d/aws-java-sdk-1.8.2.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/commons-logging/commons-logging/1.1.1/5043bfebc3db072ed80fbd362e7caf00e885d8ae/commons-logging-1.1.1.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.apache.httpcomponents/httpcore/4.2/9e431c80f58ec65460fcbdaaad802af668bb3fda/httpcore-4.2.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/commons-codec/commons-codec/1.6/b7f0fc8f61ecadeb3695f0b9464755eee44374d4/commons-codec-1.6.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.apache.httpcomponents/httpclient/4.2/894b77b74ac06206075bccc22868ef83d69e383b/httpclient-4.2.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/joda-time/joda-time/2.5/c73038a3688525aad5cf33409df483178290cd64/joda-time-2.5.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/cascading/cascading-core/2.6.1/24db1e24fea43f38d0d71496deaac2b40c9df2e9/cascading-core-2.6.1.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.codehaus.janino/commons-compiler/2.7.5/86b12400e53d73446b11eec7fbde4f099adc3057/commons-compiler-2.7.5.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/org.codehaus.janino/janino/2.7.5/bea838734aeae6e66523a7dc4eaa212c7a72da9f/janino-2.7.5.jar:/Users/luis.casillas/.gradle/caches/modules-2/files-2.1/cascading/cascading-local/2.6.1/b2be42e5e2ffbb0df68fdca50ab69968b4e1b7f6/cascading-local-2.6.1.jar" com.intellij.rt.execution.application.AppMain com.intellij.rt.execution.junit.JUnitStarter -ideVersion5 com.progressfin.cascading.misc.TypeErrorTestCase
[main] INFO cascading.property.AppProps - using app.id: 2724014C3A504111BD02B93A3B3CD329
[flow] INFO cascading.util.Version - Concurrent, Inc - Cascading 2.6.1
[flow] INFO cascading.flow.Flow - [] starting
[flow] INFO cascading.flow.Flow - []  source: FileTap["TextDelimited[['id', 'foo', 'date' | long, String, DATE]]"]["src/test/resources/TypeErrorTestCase.csv"]
[flow] INFO cascading.flow.Flow - []  sink: FileTap["TextDelimited[['id', 'bar', 'date' | long, String, DATE]]"]["build/test/output/BadAssembly.csv"]
[flow] INFO cascading.flow.Flow - []  parallel execution is enabled: true
[flow] INFO cascading.flow.Flow - []  starting jobs: 1
[flow] INFO cascading.flow.Flow - []  allocating threads: 1
[pool-1-thread-1] INFO cascading.flow.FlowStep - [] starting step: local
[pool-3-thread-1] ERROR cascading.flow.stream.TrapHandler - caught Throwable, no trap available, rethrowing
cascading.pipe.OperatorException: [pipe][com.progressfin.cascading.misc.TypeErrorTestCase$BadAssembly.<init>(TypeErrorTestCase.java:62)] operator Each failed executing operation
	at cascading.flow.stream.FilterEachStage.receive(FilterEachStage.java:81)
	at cascading.flow.stream.FilterEachStage.receive(FilterEachStage.java:34)
	at cascading.flow.stream.FunctionEachStage$1.collect(FunctionEachStage.java:80)
	at cascading.tuple.TupleEntryCollector.safeCollect(TupleEntryCollector.java:145)
	at cascading.tuple.TupleEntryCollector.add(TupleEntryCollector.java:133)
	at cascading.operation.Identity$2.operate(Identity.java:137)
	at cascading.operation.Identity.operate(Identity.java:150)
	at cascading.flow.stream.FunctionEachStage.receive(FunctionEachStage.java:99)
	at cascading.flow.stream.FunctionEachStage.receive(FunctionEachStage.java:39)
	at cascading.flow.stream.SourceStage.map(SourceStage.java:102)
[flow] INFO cascading.flow.Flow - [] stopping all jobs
	at cascading.flow.stream.SourceStage.call(SourceStage.java:53)
[flow] INFO cascading.flow.FlowStep - [] stopping: local
[flow] INFO cascading.flow.Flow - [] stopped all jobs
	at cascading.flow.stream.SourceStage.call(SourceStage.java:38)
	at java.util.concurrent.FutureTask.run(FutureTask.java:262)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
	at java.lang.Thread.run(Thread.java:744)
Caused by: java.lang.ClassCastException: java.lang.Integer cannot be cast to java.sql.Date
	at com.progressfin.cascading.misc.TypeErrorTestCase$DateFilter.isRemove(TypeErrorTestCase.java:85)
	at cascading.flow.stream.FilterEachStage.receive(FilterEachStage.java:70)
	... 15 more
[pool-3-thread-1] ERROR cascading.flow.stream.TrapHandler - caught Throwable, no trap available, rethrowing
cascading.pipe.OperatorException: [pipe][com.progressfin.cascading.misc.TypeErrorTestCase$BadAssembly.<init>(TypeErrorTestCase.java:62)] operator Each failed executing operation
	at cascading.flow.stream.FilterEachStage.receive(FilterEachStage.java:81)
	at cascading.flow.stream.FilterEachStage.receive(FilterEachStage.java:34)
	at cascading.flow.stream.FunctionEachStage$1.collect(FunctionEachStage.java:80)
	at cascading.tuple.TupleEntryCollector.safeCollect(TupleEntryCollector.java:145)
	at cascading.tuple.TupleEntryCollector.add(TupleEntryCollector.java:133)
	at cascading.operation.Identity$2.operate(Identity.java:137)
	at cascading.operation.Identity.operate(Identity.java:150)
	at cascading.flow.stream.FunctionEachStage.receive(FunctionEachStage.java:99)
	at cascading.flow.stream.FunctionEachStage.receive(FunctionEachStage.java:39)
	at cascading.flow.stream.SourceStage.map(SourceStage.java:102)
	at cascading.flow.stream.SourceStage.call(SourceStage.java:53)
	at cascading.flow.stream.SourceStage.call(SourceStage.java:38)
	at java.util.concurrent.FutureTask.run(FutureTask.java:262)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
	at java.lang.Thread.run(Thread.java:744)
Caused by: java.lang.ClassCastException: java.lang.Integer cannot be cast to java.sql.Date
	at com.progressfin.cascading.misc.TypeErrorTestCase$DateFilter.isRemove(TypeErrorTestCase.java:85)
	at cascading.flow.stream.FilterEachStage.receive(FilterEachStage.java:70)
	... 15 more
[pool-3-thread-1] ERROR cascading.flow.stream.SourceStage - caught throwable
cascading.pipe.OperatorException: [pipe][com.progressfin.cascading.misc.TypeErrorTestCase$BadAssembly.<init>(TypeErrorTestCase.java:62)] operator Each failed executing operation
	at cascading.flow.stream.FilterEachStage.receive(FilterEachStage.java:81)
	at cascading.flow.stream.FilterEachStage.receive(FilterEachStage.java:34)
	at cascading.flow.stream.FunctionEachStage$1.collect(FunctionEachStage.java:80)
	at cascading.tuple.TupleEntryCollector.safeCollect(TupleEntryCollector.java:145)
	at cascading.tuple.TupleEntryCollector.add(TupleEntryCollector.java:133)
	at cascading.operation.Identity$2.operate(Identity.java:137)
	at cascading.operation.Identity.operate(Identity.java:150)
	at cascading.flow.stream.FunctionEachStage.receive(FunctionEachStage.java:99)
	at cascading.flow.stream.FunctionEachStage.receive(FunctionEachStage.java:39)
	at cascading.flow.stream.SourceStage.map(SourceStage.java:102)
	at cascading.flow.stream.SourceStage.call(SourceStage.java:53)
	at cascading.flow.stream.SourceStage.call(SourceStage.java:38)
	at java.util.concurrent.FutureTask.run(FutureTask.java:262)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
	at java.lang.Thread.run(Thread.java:744)
Caused by: java.lang.ClassCastException: java.lang.Integer cannot be cast to java.sql.Date
	at com.progressfin.cascading.misc.TypeErrorTestCase$DateFilter.isRemove(TypeErrorTestCase.java:85)
	at cascading.flow.stream.FilterEachStage.receive(FilterEachStage.java:70)
	... 15 more

cascading.flow.FlowException: local step failed
	at cascading.flow.planner.FlowStepJob.blockOnJob(FlowStepJob.java:230)
	at cascading.flow.planner.FlowStepJob.start(FlowStepJob.java:150)
	at cascading.flow.planner.FlowStepJob.call(FlowStepJob.java:124)
	at cascading.flow.planner.FlowStepJob.call(FlowStepJob.java:43)
	at java.util.concurrent.FutureTask.run(FutureTask.java:262)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
	at java.lang.Thread.run(Thread.java:744)
Caused by: cascading.pipe.OperatorException: [pipe][com.progressfin.cascading.misc.TypeErrorTestCase$BadAssembly.<init>(TypeErrorTestCase.java:62)] operator Each failed executing operation
	at cascading.flow.stream.FilterEachStage.receive(FilterEachStage.java:81)
	at cascading.flow.stream.FilterEachStage.receive(FilterEachStage.java:34)
	at cascading.flow.stream.FunctionEachStage$1.collect(FunctionEachStage.java:80)
	at cascading.tuple.TupleEntryCollector.safeCollect(TupleEntryCollector.java:145)
	at cascading.tuple.TupleEntryCollector.add(TupleEntryCollector.java:133)
	at cascading.operation.Identity$2.operate(Identity.java:137)
	at cascading.operation.Identity.operate(Identity.java:150)
	at cascading.flow.stream.FunctionEachStage.receive(FunctionEachStage.java:99)
	at cascading.flow.stream.FunctionEachStage.receive(FunctionEachStage.java:39)
	at cascading.flow.stream.SourceStage.map(SourceStage.java:102)
	at cascading.flow.stream.SourceStage.call(SourceStage.java:53)
	at cascading.flow.stream.SourceStage.call(SourceStage.java:38)
	... 4 more
Caused by: java.lang.ClassCastException: java.lang.Integer cannot be cast to java.sql.Date
	at com.progressfin.cascading.misc.TypeErrorTestCase$DateFilter.isRemove(TypeErrorTestCase.java:85)
	at cascading.flow.stream.FilterEachStage.receive(FilterEachStage.java:70)
	... 15 more

[flow] INFO cascading.flow.Flow - [] starting
[flow] INFO cascading.flow.Flow - []  source: FileTap["TextDelimited[['id', 'foo', 'date' | long, String, DATE]]"]["src/test/resources/TypeErrorTestCase.csv"]
[flow] INFO cascading.flow.Flow - []  sink: FileTap["TextDelimited[['id', 'bar', 'date' | long, String, DATE]]"]["build/test/output/GoodAssembly.csv"]
[flow] INFO cascading.flow.Flow - []  parallel execution is enabled: true
[flow] INFO cascading.flow.Flow - []  starting jobs: 1
[flow] INFO cascading.flow.Flow - []  allocating threads: 1
[pool-4-thread-1] INFO cascading.flow.FlowStep - [] starting step: local

Process finished with exit code 255

TypeErrorTestCase

package com.progressfin.cascading.misc;

import cascading.flow.Flow;
import cascading.flow.FlowDef;
import cascading.flow.FlowProcess;
import cascading.flow.local.LocalFlowConnector;
import cascading.lingual.type.SQLDateCoercibleType;
import cascading.operation.BaseOperation;
import cascading.operation.Filter;
import cascading.operation.FilterCall;
import cascading.pipe.Each;
import cascading.pipe.Pipe;
import cascading.pipe.SubAssembly;
import cascading.pipe.assembly.Rename;
import cascading.scheme.Scheme;
import cascading.scheme.local.TextDelimited;
import cascading.tap.SinkMode;
import cascading.tap.Tap;
import cascading.tap.local.FileTap;
import cascading.tuple.Fields;
import cascading.tuple.TupleEntry;
import org.junit.Test;

import java.lang.reflect.Type;
import java.sql.Date;

/**
 * @author Luis Casillas
 */
public class TypeErrorTestCase {
    private static final Type DATE_TYPE = new SQLDateCoercibleType();

    private static final Fields INCOMING =
            new Fields("id", "foo", "date").applyTypes(long.class, String.class, DATE_TYPE);

    private static final Fields OUTGOING =
            new Fields("id", "bar", "date").applyTypes(long.class, String.class, DATE_TYPE);


    /**
     * This test case fails.
     */
    @Test
    public void badAssembly() {
        Pipe pipe = new Pipe("pipe");
        pipe = new BadAssembly(pipe);

        FlowDef flowDef = new FlowDef()
                .addSource(pipe, getInputTap())
                .addTailSink(pipe, getOutputTap("BadAssembly.csv"));

        Flow flow = new LocalFlowConnector().connect(flowDef);
        flow.complete();
    }


    static class BadAssembly extends SubAssembly {
        BadAssembly(Pipe pipe) {
            setPrevious(pipe);

            pipe = new Rename(pipe, new Fields("foo"), new Fields("bar"));
            pipe = new Each(pipe, new Fields("date"),
                            new DateFilter("date", Date.valueOf("2014-05-17"), Date.valueOf("2015-12-12")));

            setTails(pipe);
        }

    }

    static class DateFilter extends BaseOperation implements Filter {
        private final Comparable dateField;
        private final Date since;
        private final Date until;

        DateFilter(Comparable dateField, Date since, Date until) {
            super(1);
            this.dateField = dateField;
            this.since = since;
            this.until = until;
        }

        @Override
        public boolean isRemove(FlowProcess flowProcess, FilterCall filterCall) {
            TupleEntry entry = filterCall.getArguments();
            Date date = (Date)entry.getObject(dateField, Date.class);  // FIXME: ClassCast exception here
            return date.before(since) || !until.before(date);
        }

    }


    /**
     * This test case passes.  Difference: the assembly used declares a type for the Rename output field.
     */
    @Test
    public void goodAssembly() {
        Pipe pipe = new Pipe("pipe");
        pipe = new GoodAssembly(pipe);

        FlowDef flowDef = new FlowDef()
                .addSource(pipe, getInputTap())
                .addTailSink(pipe, getOutputTap("GoodAssembly.csv"));

        Flow flow = new LocalFlowConnector().connect(flowDef);
        flow.complete();
    }

    static class GoodAssembly extends SubAssembly {
        GoodAssembly(Pipe pipe) {
            setPrevious(pipe);

            // The only difference is the String.class in the next line...
            pipe = new Rename(pipe, new Fields("foo"), new Fields("bar", String.class));
            pipe = new Each(pipe, new Fields("date"),
                            new DateFilter("date", Date.valueOf("2014-05-17"), Date.valueOf("2015-12-12")));

            setTails(pipe);
        }

    }


    private Tap getInputTap() {
        Scheme scheme = new TextDelimited(INCOMING, true, true, ",");
        return new FileTap(scheme, "src/test/resources/TypeErrorTestCase.csv", SinkMode.KEEP);
    }

    private Tap getOutputTap(String filename) {
        Scheme scheme = new TextDelimited(OUTGOING, true, true, ",");
        return new FileTap(scheme, "build/test/output/" + filename, SinkMode.REPLACE);
    }
}

TypeErrorTestCase.csv